Realtime Compute for Apache Flink:Desarrollo y depuración

• How should I declare a DDL statement when submitting it with DML statements in the same script?

• How do I write multiple INSERT INTO statements?

• How do I pass special characters in Entry point Main Arguments?

• Why does uploading a UDF JAR package fail after multiple modifications?

• Why are fields misaligned when a POJO class is used as the UDTF return type?

• How do I resolve Flink dependency conflicts?

• Error: Could not parse type at position 50: expected but was . Input type string: ROW

• Error when writing to a table: "Invalid primary key. Column 'xxx' is nullable."

• Why does a JSON file open in the browser instead of being downloaded?

Declaración de DDL con sentencias DML

Cuando envíe sentencias DDL y DML juntas en el mismo script, declare la sentencia DDL como CREATE TEMPORARY TABLE en lugar de CREATE TABLE. De lo contrario, al hacer clic en Validate se producirá un error similar al siguiente.

CREATE TABLE datagen_source (a bigint, b int, c varchar)
  WITH ('connector' = 'datagen');
CREATE  TABLE print_sink(C bigint, var1 int)
  WITH ('connector' = 'print','logger' = 'true');
INSERT INTO print_sink SELECT a,8  FROM datagen_source;
Error message:
org.apache.flink.table.gateway.api.vvr.utils.SqlValidationException: A sequence of multiple statements to execute is supported if the last statement is a 'SELECT' statement or 'INSERT INTO' statement or 'CREATE TABLE IF NOT EXISTS ... AS TABLE' statement ...BASE IF NOT EXISTS ... AS DATABASE' statement or 'AUTO OPTIMIZE TABLE|DATABASE' statements or multiple 'INSERT INTO' or 'CREATE TABLE IF NOT EXISTS ... AS TAB ...ATABASE IF NOT EXISTS ... AS DATABASE' statements wrapped in a 'BEGIN STATEMENT SET' block and all other statements are CREATE TEMPORARY TABLE|VIEW|[SYSTEM] FUNCTION, 'SHOW', DESCRIBE, 'USE' statements.
	at org.apache.flink.table.gateway.vvr.service.utils.SqlValidateUtils.validateDraft(SqlValidateUtils.java:107)
	at org.apache.flink.table.gateway.vvr.service.command.DraftCommand.getDraftType(DraftCommand.java:120)
	at org.apache.flink.table.gateway.vvr.service.command.DraftCommand.executeInternal(DraftCommand.java:71)
	at java.security.AccessController.doPrivileged(Native Method)
	at javax.security.auth.Subject.doAs(Subject.java:422)

Varias sentencias INSERT INTO

Para formar una unidad lógica única, agrupe varias sentencias INSERT INTO entre BEGIN STATEMENT SET; y END;. Para obtener más información, consulte INSERT INTO statements. Si no agrupa las sentencias, al hacer clic en Validate se producirá un error similar al siguiente.

CREATE TEMPORARY TABLE datagen_source (a bigint, b int, c varchar)
  WITH ('connector' = 'datagen');
CREATE TEMPORARY  TABLE print_sink(C bigint, var1 int)
  WITH ('connector' = 'print','logger' = 'true');
CREATE TEMPORARY TABLE print_sink2(C bigint, var2 int)
  WITH ('connector' = 'print','logger' = 'true');
INSERT INTO print_sink SELECT a,B  FROM datagen_source;
INSERT INTO print_sink2 SELECT a,B  FROM datagen_source;
org.apache.flink.table.gateway.api.vvr.utils.SqlValidationException: A sequence of multiple statements to execute is supported if the last statement is a 'SELECT' statement or 'INSERT INTO' statement or 'CREATE TABLE IF NOT EXISTS ... AS TABLE' statement or 'CREATE DATABASE IF NOT EXISTS ... AS DATABASE' statement or 'AUTO OPTIMIZE TABLE|DATABASE' statements or multiple 'INSERT INTO' or 'CREATE TABLE IF NOT EXISTS ... AS TABLE' or 'CREATE DATABASE IF NOT EXISTS ... AS DATABASE' statements wrapped in a 'BEGIN STATEMENT SET' block and all other statements are CREATE TEMPORARY TABLE|VIEW|[SYSTEM] FUNCTION, 'SHOW', DESCRIBE, 'USE' statements.
      at org.apache.flink.table.gateway.vvr.service.utils.SqlValidateUtils.validateDraft(SqlValidateUtils.java:79)
      at org.apache.flink.table.gateway.vvr.service.command.DraftCommand.getDraftType(DraftCommand.java:120)
      at org.apache.flink.table.gateway.vvr.service.command.DraftCommand.executeInternal(DraftCommand.java:71)
      at java.security.AccessController.doPrivileged(Native Method)
      at javax.security.auth.Subject.doAs(Subject.java:422)
      at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1899)
      at org.apache.flink.table.gateway.service.context.SqlGatewaySecurityContext.runSecured(SqlGatewaySecurityContext.java:73)
      at org.apache.flink.table.gateway.vvr.service.command.AbstractCommand.wrapClassLoader(AbstractCommand.java:171)
      at org.apache.flink.table.gateway.vvr.service.command.AbstractCommand.execute(AbstractCommand.java:163)
      at org.apache.flink.table.gateway.vvr.service.command.CommandManager.lambda$execute$0(CommandManager.java:71)
      at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
      at java.util.concurrent.FutureTask.run(FutureTask.java:266)
      at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)

Paso de caracteres especiales en argumentos principales

• Causa

  Cuando pasa caracteres especiales como # y $ en Entry point Main Arguments, el carácter de escape de barra invertida (\) no funciona y los caracteres especiales se descartan.

• Solución

  En la página Deployments, haga clic en el nombre del despliegue objetivo. En la sección Parameters, agregue el parámetro env.java.opts: -Dconfig.disable-inline-comment=true al campo Other Configuration. Para obtener más información, consulte How to configure custom deployment parameters.

Error de carga del JAR de UDF tras modificación

• Causa

  El entorno de ejecución de UDF exige nombres de clase únicos en todos los paquetes JAR.

• Solución

  • Elimine el paquete JAR anterior y cargue el nuevo.

  • Cargue el paquete JAR en la sección Additional Dependencies y utilice una función temporal en su código. Para obtener información sobre cómo usar una función temporal, consulte Register a UDF. El siguiente ejemplo muestra la sintaxis.

    CREATE TEMPORARY FUNCTION `cp_record_reduce` AS 'com.taobao.test.udf.blink.CPRecordReduceUDF';

    En la sección Additional Dependencies, ubicada en el panel Other Configuration a la derecha, proporcione la URL de OSS de su paquete JAR de UDF.

Desalineación de campos con clase POJO como tipo de retorno de UDTF****

• Detalles

  Cuando utiliza una clase POJO como tipo de retorno para una UDTF y declara explícitamente una lista de alias para las columnas devueltas en SQL, puede encontrar un problema de desalineación de campos. Como resultado, es posible que los campos reales no sean los esperados, incluso si los tipos de datos son consistentes.

  Por ejemplo, si utiliza la siguiente clase POJO como tipo de retorno para una UDTF, la empaqueta y registra la función como una deployment-level UDF tal como se describe en Develop a custom function, la validación de SQL fallará.

  package com.aliyun.example;
  public class TestPojoWithoutConstructor {
  	public int c;
  	public String d;
  	public boolean a;
  	public String b;
  }

  package com.aliyun.example;
  import org.apache.flink.table.functions.TableFunction;
  public class MyTableFuncPojoWithoutConstructor extends TableFunction<TestPojoWithoutConstructor> {
  	private static final long serialVersionUID = 1L;
  	public void eval(String str1, Integer i2) {
  		TestPojoWithoutConstructor p = new TestPojoWithoutConstructor();
  		p.d = str1 + "_d";
  		p.c = i2 + 2;
  		p.b = str1 + "_b";
  		collect(p);
  	}
  }

  CREATE TEMPORARY FUNCTION MyTableFuncPojoWithoutConstructor as 'com.aliyun.example.MyTableFuncPojoWithoutConstructor';
  CREATE TEMPORARY TABLE src ( 
    id STRING,
    cnt INT
  ) WITH (
    'connector' = 'datagen'
  );
  CREATE TEMPORARY TABLE sink ( 
    f1 INT,
    f2 STRING,
    f3 BOOLEAN,
    f4 STRING
  ) WITH (
   'connector' = 'print'
  );
  INSERT INTO sink
  SELECT T.* FROM src, LATERAL TABLE(MyTableFuncPojoWithoutConstructor(id, cnt)) AS T(c, d, a, b);

  La validación de SQL devuelve el siguiente mensaje de error:

  org.apache.flink.table.api.ValidationException: SQL validation failed. Column types of query result and sink for 'vvp.default.sink' do not match.
  Cause: Sink column 'f1' at position 0 is of type INT but expression in the query is of type BOOLEAN NOT NULL.
  Hint: You will need to rewrite or cast the expression.
  Query schema: [c: BOOLEAN NOT NULL, d: STRING, a: INT NOT NULL, b: STRING]
  Sink schema:  [f1: INT, f2: STRING, f3: BOOLEAN, f4: STRING]
  	at org.apache.flink.table.sqlserver.utils.FormatValidatorExceptionUtils.newValidationException(FormatValidatorExceptionUtils.java:41)

  Los campos devueltos por la UDTF están desalineados respecto a los campos de la clase POJO. En el resultado de la consulta, el campo c es de tipo BOOLEAN y el campo a es de tipo INT, lo cual es inverso a sus definiciones en la clase POJO.

• Causa

  De acuerdo con las reglas de inferencia de tipos para clases POJO:

  • Si la clase POJO tiene un constructor parametrizado, Flink infiere el tipo de retorno según el orden de los parámetros del constructor.

  • Si la clase POJO no tiene un constructor parametrizado, Flink reordena los campos en orden alfabético por nombre.

  En el ejemplo, dado que la clase POJO utilizada para el tipo de retorno de la UDTF carece de un constructor parametrizado, los campos se devuelven en orden alfabético, lo que resulta en el tipo BOOLEAN a, VARCHAR(2147483647) b, INTEGER c, VARCHAR(2147483647) d). Aunque esta inferencia es válida, la consulta SQL renombra explícitamente las columnas de salida con LATERAL TABLE(MyTableFuncPojoWithoutConstructor(id, cnt)) AS T(c, d, a, b). Esta lista de alias cambia el nombre de las columnas por posición, lo que crea una discrepancia con los campos ordenados alfabéticamente desde el POJO. Este conflicto entre el uso de alias posicionales y el ordenamiento alfabético de campos provoca la excepción de validación o una desalineación inesperada de los datos.

• Solución

  • Si la clase POJO carece de un constructor parametrizado, elimine el cambio de nombre explícito de los campos de retorno de la UDTF. Por ejemplo, cambie la sentencia INSERT en el SQL a:

    -- If the POJO class lacks a parameterized constructor, select the required fields by name. 
    -- When using T.*, you must be aware of the actual order of the returned fields.
    SELECT T.c, T.d, T.a, T.b FROM src, LATERAL TABLE(MyTableFuncPojoWithoutConstructor(id, cnt)) AS T;

  • Implemente un constructor parametrizado en la clase POJO para controlar el orden de los campos en el tipo de retorno. En este caso, el orden de los campos de la salida de la UDTF coincidirá con el orden de los parámetros del constructor.

    package com.aliyun.example;
    public class TestPojoWithConstructor {
    	public int c;
    	public String d;
    	public boolean a;
    	public String b;
    	// Using specific fields order instead of alphabetical order
    	public TestPojoWithConstructor(int c, String d, boolean a, String b) {
    		this.c = c;
    		this.d = d;
    		this.a = a;
    		this.b = b;
    	}
    }

Resolución de conflictos de dependencias de Flink

• Síntomas

  • El conflicto se manifiesta como errores claros arrojados por clases relacionadas con Flink o Hadoop.

    java.lang.AbstractMethodError
    java.lang.ClassNotFoundException
    java.lang.IllegalAccessError
    java.lang.IllegalAccessException
    java.lang.InstantiationError
    java.lang.InstantiationException
    java.lang.InvocationTargetException
    java.lang.NoClassDefFoundError
    java.lang.NoSuchFieldError
    java.lang.NoSuchFieldException
    java.lang.NoSuchMethodError
    java.lang.NoSuchMethodException

  • Alternativamente, el problema puede presentarse como un comportamiento inesperado sin un mensaje de error claro, por ejemplo:

    • No se generan registros o la configuración de log4j no surte efecto.

      Este problema suele deberse a configuraciones relacionadas con log4j incluidas en las dependencias. Compruebe si el paquete JAR del despliegue contiene dependencias que incluyan configuraciones de log4j. Puede eliminar estas configuraciones mediante exclusiones en sus definiciones de dependencias.

      Nota

      Si debe usar una versión diferente de log4j, utilice maven-shade-plugin para reubicar las clases relacionadas con log4j.

    • Excepciones en llamadas RPC.

      Los conflictos de dependencias que afectan las llamadas RPC de Akka de Flink pueden causar excepciones que no se muestran en los registros de forma predeterminada. Debe habilitar el registro de depuración para identificarlas.

      Por ejemplo, el registro de depuración muestra Cannot allocate the requested resources. Trying to allocate ResourceProfile{xxx}, pero el registro del JobManager (JM) no muestra actividad después de Registering TaskManager with ResourceID xxx hasta que ocurre un error de tiempo de espera NoResourceAvailableException. Mientras tanto, el TaskManager (TM) reporta continuamente el error Cannot allocate the requested resources. Trying to allocate ResourceProfile{xxx}.

      Causa: Con el registro de depuración habilitado, se observa que se arroja una InvocationTargetException durante una llamada RPC. Este error provoca que la asignación de slots del TM falle a mitad del proceso, lo que resulta en un estado inconsistente. Posteriormente, el ResourceManager (RM) intenta asignar un slot de forma continua y sin éxito, y no logra recuperarse.

• Causas

  • El paquete JAR del despliegue contiene dependencias innecesarias, como bibliotecas base de Flink, Hadoop o log4j, lo que provoca conflictos de dependencias.

  • Las dependencias para un conector requerido no están incluidas en el paquete JAR.

• Solución de problemas

  • Revise el archivo pom.xml del despliegue en busca de dependencias innecesarias.

  • Inspeccione el contenido del paquete JAR del despliegue ejecutando el comando jar tf foo.jar para buscar archivos en conflicto.

  • Analice el árbol de dependencias del despliegue en busca de conflictos ejecutando el comando mvn dependency:tree.

• Solución

  • Como práctica recomendada, establezca el scope de las dependencias del framework base en provided. Esto evita que se incluyan en el paquete JAR del despliegue.

    • DataStream Java

      <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java_2.11</artifactId>
        <version>${flink.version}</version>
        <scope>provided</scope>
      </dependency>

    • DataStream Scala

      <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-scala_2.11</artifactId>
        <version>${flink.version}</version>
        <scope>provided</scope>
      </dependency>

    • DataSet Java

      <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>${flink.version}</version>
        <scope>provided</scope>
      </dependency>

    • DataSet Scala

      <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-scala_2.11</artifactId>
        <version>${flink.version}</version>
        <scope>provided</scope>
      </dependency>

  • Agregue las dependencias del conector requerido a su proyecto. El scope predeterminado es compile, lo cual las incluye correctamente en el paquete JAR del despliegue. Por ejemplo, para agregar el conector de Kafka:

    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-kafka_2.11</artifactId>
        <version>${flink.version}</version>
    </dependency>

  • No agregue otras dependencias de Flink, Hadoop o log4j. Sin embargo:

    • Si el despliegue tiene una dependencia directa de componentes de configuración base o relacionados con conectores, establezca el scope en provided. El siguiente ejemplo muestra la sintaxis.

      <dependency>
          <groupId>org.apache.hadoop</groupId>
          <artifactId>hadoop-common</artifactId>
          <scope>provided</scope>
      </dependency>

    • Si el despliegue tiene una dependencia transitiva de componentes de configuración base o relacionados con conectores, elimínela mediante una exclusion. El siguiente ejemplo muestra la sintaxis.

      <dependency>
          <groupId>foo</groupId>
            <artifactId>bar</artifactId>
            <exclusions>
              <exclusion>
              <groupId>org.apache.hadoop</groupId>
              <artifactId>hadoop-common</artifactId>
             </exclusion>
          </exclusions>
      </dependency>

Error: Could not parse type at position 50: expected but was . Input type string: ROW

• Detalles del error

  Al escribir SQL en el editor de SQL, se produce un error de verificación de sintaxis (línea ondulada roja) cuando utiliza una UDTF.

  Caused by: org.apache.flink.table.api.ValidationException: Could not parse type at position 50: <IDENTIFIER> expected but was <KEYWORD>. Input type string: ROW<resultId String,pointRange String,from String,to String,type String,pointScope String,userId String,point String,triggerSource String,time String,uuid String>

  El siguiente código es un ejemplo:

  @FunctionHint(
      //input = @DataTypeHint("BYTES"),
      output = @DataTypeHint("ROW<resultId String,pointRange String,from String,to String,type String,pointScope String,userId String,point String,triggerSource String,time String,uuid String>"))
  public class PointChangeMetaQPaser1 extends TableFunction<Row> {
      Logger logger = LoggerFactory.getLogger(this.getClass().getName());
      public void eval(byte[] bytes) {
          try {
              String messageBody = new String(bytes, "UTF-8");
              Map<String, String> resultDO = JSON.parseObject(messageBody, Map.class);
              logger.info("PointChangeMetaQPaser1 logger:" + JSON.toJSONString(resultDO));
              collect(Row.of(
                      getString(resultDO.get("resultId")),
                      getString(resultDO.get("pointRange")),
                      getString(resultDO.get("from")),
                      getString(resultDO.get("to")),
                      getString(resultDO.get("type")),
                      getString(resultDO.get("pointScope")),
                      getString(resultDO.get("userId")),
                      getString(resultDO.get("point")),
                      getString(resultDO.getOrDefault("triggerSource", "NULL")),
                      getString(resultDO.getOrDefault("time", String.valueOf(System.currentTimeMillis()))),
                      getString(resultDO.getOrDefault("uuid", String.valueOf(UUID.randomUUID())))
              ));
          } catch (Exception e) {
              logger.error("PointChangeMetaQPaser1 error", e);
          }
      }
      private String getString(Object o) {
          if (o == null) {
              return null;
          }
          return String.valueOf(o);
      }
  }

• Causa

  Cuando utiliza DataTypeHint para definir tipos de datos de funciones, se usa directamente una palabra clave reservada como nombre de campo.

• Solución

  • Cambie el nombre de los campos a términos que no sean palabras clave. Por ejemplo, cambie to por fto y from por ffrom.

  • Encierre entre comillas invertidas () los campos que utilicen palabras clave reservadas.

Error: "Invalid primary key. Column 'xxx' is nullable."

• Causa

  Flink exige que todas las columnas de clave primaria se declaren explícitamente como NOT NULL. Incluso si los datos no contienen valores NULL, Flink rechaza la operación antes de escribir si una columna de clave primaria en la sentencia de creación de tabla admite valores NULL (por ejemplo, INT NULL). Esto no es un error de tiempo de ejecución, sino una comprobación semántica durante la fase de análisis de DDL.

• Solución

  Declare las columnas de clave primaria mencionadas en el error como NOT NULL y vuelva a crear la tabla.

El archivo JSON se abre en el navegador en lugar de descargarse

• Síntoma

  Cuando hace clic para descargar un archivo JSON desde la página Artifacts, el navegador no inicia la descarga. En su lugar, abre una nueva pestaña y muestra el contenido JSON directamente.

• Causa

  Al archivo JSON en OSS le falta el encabezado de respuesta HTTP Content-Disposition: attachment. Esto provoca que el navegador muestre el contenido del archivo directamente en lugar de descargarlo.

• Solución

  • Opción 1: Volver a cargar el archivo

    Este problema se corrigió en la versión 4.5.0 de la plataforma, pero la corrección solo aplica a los archivos cargados después del lanzamiento de esta versión. Los archivos cargados antes de esta fecha deben manejarse manualmente.

  • Opción 2: Modificar los metadatos del objeto OSS

    Modifique manualmente los metadatos del objeto agregando el siguiente atributo HTTP estándar:

    • Nombre del encabezado: Content-Disposition

    • Valor del encabezado: attachment

    Para obtener más información, consulte Manage object metadata.