Realtime Compute for Apache Flink:Desenvolvimento e depuração

• Como declarar uma instrução DDL ao enviá-la com instruções DML no mesmo script?

• Como escrever várias instruções INSERT INTO?

• Como passar caracteres especiais em Entry point Main Arguments?

• Por que o upload de um pacote JAR UDF falha após várias modificações?

• Por que os campos ficam desalinhados ao usar uma classe POJO como tipo de retorno UDTF?

• Como resolver conflitos de dependência do Flink?

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

• Erro ao gravar em uma tabela: "Invalid primary key. Column 'xxx' is nullable."

• Por que um arquivo JSON abre no navegador em vez de ser baixado?

Declaração de DDL com instruções DML

Ao enviar instruções DDL e DML juntas no mesmo script, declare a instrução DDL como CREATE TEMPORARY TABLE em vez de CREATE TABLE. Caso contrário, a validação em Validate falhará com um erro semelhante ao seguinte.

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)

Várias instruções INSERT INTO

Para formar uma única unidade lógica, envolva várias instruções INSERT INTO entre BEGIN STATEMENT SET; e END;. Para obter mais informações, consulte Instruções INSERT INTO. Sem esse encapsulamento, a validação em Validate falhará com um erro semelhante ao seguinte.

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)

Passagem de caracteres especiais nos argumentos principais

• Causa

  Ao passar caracteres especiais como # e $ em Entry point Main Arguments, o caractere de escape barra invertida (\) não funciona e os caracteres especiais são descartados.

• Solução

  Na página Deployments, clique em no nome da deployment desejada. Na seção Parameters, adicione o parâmetro env.java.opts: -Dconfig.disable-inline-comment=true ao campo Other Configuration. Para obter mais informações, consulte Como configurar parâmetros personalizados de deployment.

Falha no upload de JAR UDF após modificação

• Causa

  O runtime UDF exige nomes de classe exclusivos entre pacotes JAR.

• Solução

  • Exclua o pacote JAR antigo e faça o upload do novo.

  • Faça o upload do pacote JAR na seção Additional Dependencies e use uma função temporária no código. Para obter informações sobre como usar uma função temporária, consulte Registrar uma UDF. O exemplo a seguir mostra a sintaxe.

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

    Na seção Additional Dependencies, localizada no painel Other Configuration à direita, forneça a URL do OSS do seu pacote JAR UDF.

Desalinhamento de campos com classe POJO como tipo de retorno UDTF****

• Detalhes

  Ao usar uma classe POJO como tipo de retorno para uma UDTF e declarar explicitamente uma lista de aliases para as colunas retornadas em SQL, pode ocorrer um problema de desalinhamento de campos. Consequentemente, os campos reais podem não corresponder aos esperados, mesmo que os tipos de dados sejam consistentes.

  Por exemplo, se você usar a seguinte classe POJO como tipo de retorno para uma UDTF, empacotá-la e registrar a função como uma UDF no nível de deployment conforme descrito em Desenvolver uma função personalizada, a validação SQL falhará.

  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);

  A validação SQL retorna a seguinte mensagem de erro:

  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)

  Os campos retornados da UDTF estão desalinhados em relação aos campos da classe POJO. No resultado da consulta, o campo c é do tipo BOOLEAN e o campo a é do tipo INT, o inverso de suas definições na classe POJO.

• Causa

  De acordo com as regras de inferência de tipo para classes POJO:

  • Se a classe POJO tiver um construtor parametrizado, o Flink infere o tipo de retorno com base na ordem dos parâmetros do construtor.

  • Se a classe POJO não tiver um construtor parametrizado, o Flink reordena os campos em ordem alfabética por nome.

  No exemplo, como a classe POJO usada para o tipo de retorno da UDTF não possui um construtor parametrizado, os campos são retornados em ordem alfabética, resultando no tipo BOOLEAN a, VARCHAR(2147483647) b, INTEGER c, VARCHAR(2147483647) d). Embora essa inferência seja válida, a consulta SQL renomeia explicitamente as colunas de saída com LATERAL TABLE(MyTableFuncPojoWithoutConstructor(id, cnt)) AS T(c, d, a, b). Essa lista de aliases renomeia as colunas por posição, criando uma incompatibilidade com os campos ordenados alfabeticamente da POJO. Esse conflito entre alias posicional e ordenação alfabética de campos causa a exceção de validação ou desalinhamento inesperado de dados.

• Solução

  • Se a classe POJO não tiver um construtor parametrizado, remova a renomeação explícita dos campos de retorno da UDTF. Por exemplo, altere a instrução INSERT no SQL para:

    -- 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 um construtor parametrizado na classe POJO para controlar a ordem dos campos no tipo de retorno. Nesse caso, a ordem dos campos da saída da UDTF corresponderá à ordem dos parâmetros do construtor.

    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;
    	}
    }

Resolução de conflitos de dependência do Flink

• Sintomas

  • O conflito se manifesta como erros claros lançados por classes relacionadas ao Flink ou 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, o problema pode se apresentar como um comportamento inesperado sem uma mensagem de erro clara, como:

    • Logs não são gerados ou a configuração do log4j não entra em vigor.

      Esse problema geralmente é causado por configurações relacionadas ao log4j incluídas nas dependências. Verifique se o pacote JAR da deployment contém dependências que carregam configurações do log4j. Remova essas configurações usando exclusões nas definições de dependência.

      Nota

      Se for necessário usar uma versão diferente do log4j, use o maven-shade-plugin para realocar as classes relacionadas ao log4j.

    • Exceções de chamada RPC.

      Conflitos de dependência que afetam as chamadas Akka RPC do Flink podem causar exceções não exibidas nos logs por padrão. Ative o log de depuração para identificá-las.

      Por exemplo, o log de depuração mostra Cannot allocate the requested resources. Trying to allocate ResourceProfile{xxx}, mas o log do JobManager (JM) não mostra atividade após Registering TaskManager with ResourceID xxx até ocorrer um erro de tempo limite NoResourceAvailableException. Enquanto isso, o TaskManager (TM) relata continuamente o erro Cannot allocate the requested resources. Trying to allocate ResourceProfile{xxx}.

      Causa: Com o log de depuração ativado, é possível ver que uma InvocationTargetException é lançada durante uma chamada RPC. Esse erro faz com que a alocação de slot do TM falhe no meio do processo, resultando em um estado inconsistente. O ResourceManager (RM) tenta então contínua e malsucedidamente alocar um slot, sem conseguir se recuperar.

• Causas

  • O pacote JAR da deployment contém dependências desnecessárias, como bibliotecas base do Flink, Hadoop ou log4j, que causam conflitos de dependência.

  • Dependências para um conector necessário não estão incluídas no pacote JAR.

• Solução de problemas

  • Revise o arquivo pom.xml da deployment em busca de dependências desnecessárias.

  • Inspecione o conteúdo do pacote JAR da deployment executando o comando jar tf foo.jar para verificar arquivos conflitantes.

  • Analise a árvore de dependências da deployment em busca de conflitos executando o comando mvn dependency:tree.

• Solução

  • Como prática recomendada, defina o scope das dependências básicas do framework como provided. Isso impede que elas sejam agrupadas no pacote JAR da deployment.

    • 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>

  • Adicione as dependências de conector necessárias ao seu projeto. O escopo padrão é compile, que as agrupa corretamente no pacote JAR da deployment. Por exemplo, para adicionar o conector Kafka:

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

  • Não adicione outras dependências do Flink, Hadoop ou log4j. No entanto:

    • Se a deployment tiver uma dependência direta de componentes de configuração base ou relacionados a conectores, defina o escopo como provided. O exemplo a seguir mostra a sintaxe.

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

    • Se a deployment tiver uma dependência transitiva de componentes de configuração base ou relacionados a conectores, remova-a usando uma exclusion. O exemplo a seguir mostra a sintaxe.

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

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

• Detalhes do erro

  Ao escrever SQL no editor SQL, ocorre um erro de verificação de sintaxe (linha ondulada vermelha) ao usar uma 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>

  O código a seguir é um exemplo:

  @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

  Ao usar DataTypeHint para definir tipos de dados de função, uma palavra-chave reservada é usada diretamente como nome de campo.

• Solução

  • Renomeie os campos para termos que não sejam palavras-chave. Por exemplo, renomeie to para fto e from para ffrom.

  • Envolva campos que usam palavras-chave reservadas em crases ().

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

• Causa

  O Flink exige que todas as colunas de chave primária sejam declaradas explicitamente como NOT NULL. Mesmo que os dados não contenham valores NULL, o Flink rejeita a operação antes da gravação se uma coluna de chave primária na instrução de criação da tabela permitir valores NULL (por exemplo, INT NULL). Este não é um erro de runtime, mas uma verificação semântica durante a fase de análise DDL.

• Solução

  Declare as colunas de chave primária mencionadas no erro como NOT NULL e recrie a tabela.

Arquivo JSON abre no navegador em vez de ser baixado

• Sintoma

  Ao clicar em para baixar um arquivo JSON na página Artifacts, o navegador não inicia o download. Em vez disso, ele abre uma nova aba e exibe o conteúdo JSON diretamente.

• Causa

  O arquivo JSON no OSS está sem o cabeçalho de resposta HTTP Content-Disposition: attachment. Isso faz com que o navegador exiba o conteúdo do arquivo diretamente em vez de baixá-lo.

• Solução

  • Opção 1: Fazer upload do arquivo novamente

    Esse problema foi corrigido na versão 4.5.0 da plataforma, mas a correção aplica-se apenas a arquivos enviados após o lançamento dessa versão. Arquivos enviados antes dessa data devem ser tratados manualmente.

  • Opção 2: Modificar os metadados do objeto OSS

    Modifique manualmente os metadados do objeto adicionando o seguinte atributo HTTP padrão:

    • Nome do cabeçalho: Content-Disposition

    • Valor do cabeçalho: attachment

    Para obter mais informações, consulte Gerenciar metadados de objetos.