Realtime Compute for Apache Flink:Catalogs FAQ

Job do Hologres Catalog termina sem consumir dados em tempo real

Sintomas

O job do Flink é concluído após ler os dados existentes do Hologres e encerra. Ele não aguarda a chegada de novos registros.

Causa

Por padrão, o Realtime Compute for Apache Flink lê dados do Hologres em modo batch, não em modo streaming.

Solução

Adicione a dica /*+ OPTIONS('binlog'='true') */ à consulta SQL para alternar para o modo streaming:

INSERT INTO testfinish
SELECT
  col1,
  col2,
  col3
FROM testsource /*+ OPTIONS ('binlog' = 'true') */
   LEFT JION testdim /*+ OPTIONS ('binlog' = 'true') */
         ON testsource.co11 = testdim.'col1';

Erro na interface CatalogTableProvider com CREATE TABLE AS

Mensagem de erro

org.apache.flink.table.api.ValidationException: SQL validation failed. CREATE TABLE ... AS TABLE ... statement requires target catalog 'XXXXXX' (MongoDBCatalog) implements org.apache.flink.table.catalog.CatalogTableProvider interface. Hint: Please refer the document and use another catalog that supports schema evolution as the target catalog.

Causa

O catálogo da tabela sink não implementa a interface CatalogTableProvider e não pode ser usado com a instrução CREATE TABLE AS (CTAS). Por exemplo, catálogos MongoDB não suportam CTAS como sink.

Solução

Utilize uma tabela source e uma tabela sink compatíveis com CTAS. Para obter a lista completa de sistemas de armazenamento suportados, consulte CREATE TABLE AS (CTAS).

Consultas ao MySQL catalog estão lentas ou retornam exceções de servidor

Respostas lentas ou exceções de servidor ao consultar um MySQL catalog geralmente resultam de alta latência de rede ou conexões instáveis entre o Flink e o MySQL. Aplique a correção adequada ao seu cenário.

-- Author:        izha****69041.onaliyun.com
-- Created Time:  2022-06-30 14:30:42
-- Description:   Write your description here

SELECT
  order_id,
  SUM(goods_id) AS nm
from
  `bi`.`izm_tempfile`.`orders`
GROUP BY order_id

Server response error
com.ververica.cdc.connectors.shaded.com.zaxxer.hikari.pool.HikariPool$PoolInitializationException: Failed to initialize pool: Unsupported transaction isolation level '-1'
  at com.ververica.cdc.connectors.shaded.com.zaxxer.hikari.pool.HikariPool.throwPoolInitializationException(HikariPool.java:596)
  at com.ververica.cdc.connectors.shaded.com.zaxxer.hikari.pool.HikariPool.checkFailFast(HikariPool.java:575)
  at com.ververica.cdc.connectors.shaded.com.zaxxer.hikari.pool.HikariPool.(HikariPool.java:115)
  at com.ververica.cdc.connectors.shaded.com.zaxxer.hikari.HikariDataSource.(HikariDataSource.java:81)
  at com.ververica.cdc.connectors.mysql.source.connection.PooledDataSourceFactory.createPooledDataSource(PooledDataSourceFactory.java:35)
  at com.ververica.cdc.connectors.mysql.source.connection.JdbcConnectionPools.getOrCreateConnectionPool(JdbcConnectionPools.java:50)
  at com.ververica.connectors.mysql.connection.MySqlConnectionProvider.(MySqlConnectionProvider.java:56)
  at com.ververica.connectors.mysql.catalog.MySqlCatalog.open(MySqlCatalog.java:149)
  at org.apache.flink.table.catalog.CatalogManager.registerCatalog(CatalogManager.java:210)
  at org.apache.flink.table.api.internal.TableEnvironmentImpl.createCatalog(TableEnvironmentImpl.java:1320)
  at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:1161)
  at org.apache.flink.table.sqlserver.execution.OperationExecutorImpl.executeOperation(OperationExecutorImpl.java:434)
  at org.apache.flink.table.sqlserver.execution.DelegateOperationExecutor.lambda$executeOperation...

Causa 1: Largura de banda pública insuficiente

Verifique se a largura de banda pública do Realtime Compute for Apache Flink para o MySQL está muito baixa. A largura de banda pública padrão pode ser de apenas 2 Mbit/s. Confirme também se há acesso entre regiões. Para instruções sobre como visualizar sua largura de banda, consulte How do I view the public bandwidth?

Causa 2: Conexão de rede instável

Aumente os valores dos parâmetros interactive_timeout e wait_timeout na instância MySQL para reduzir a frequência de quedas de conexão.

Causa 3: Tempo limite de conexão muito curto

Inclua a opção 'connect.timeout'='120s' na cláusula WITH ao criar uma tabela para o MySQL catalog:

-- Add to the WITH clause of your CREATE TABLE statement
'connect.timeout' = '120s'

Não é possível obter token STS do meta-service do EMR

Mensagem de erro

Caused by: MetaException(message:Initialize DlfMetaStoreClient failed: Initialize DlfMetaStoreClient failed: Cannot obtain STS token from EMR meta-service. Note that AK-Mode[dlf.catalog.akMode] can only used in EMR clusters, otherwise you should config the dlf.catalog.accessKeyId and dlf.catalog.accessKeySecret explicitly.)
  at com.aliyun.datalake.metastore.hive2.ProxyMetaStoreClient.createClient(ProxyMetaStoreClient.java:91)
  at com.aliyun.datalake.metastore.hive2.ProxyMetaStoreClient.<init>(ProxyMetaStoreClient.java:71)
  ... 41 more

Causa

O arquivo de configuração hive-site.xml contém uma definição EMR_AUTO para dlf.catalog.akMode. Essa configuração funciona apenas dentro de clusters E-MapReduce (EMR). Executá-la fora do EMR causa falha na inicialização do Hive catalog.

Solução

Atualize o arquivo hive-site.xml no diretório hive-conf-dir:

1. Remova a seguinte propriedade:

   <property>
     <name>dlf.catalog.akMode</name>
     <value>EMR_AUTO</value>
   </property>

2. Adicione seu AccessKey ID e AccessKey secret:

   <property>
       <name>dlf.catalog.accessKeyId</name>
       <value>${AccessKeyId}</value>
     </property>
   <property>
       <name>dlf.catalog.accessKeySecret</name>
       <value>${AccessKeySecret}</value>
     </property>

Alterações de configuração de catálogo no console ou via ALTER CATALOG não têm efeito

Causa

• O Realtime Compute for Apache Flink não suporta a modificação da configuração de um catálogo registrado. O console não oferece entrada para tais modificações, e a instrução ALTER CATALOG não altera configurações de catálogo.

• A configuração de um catálogo de terceiros, como DLF ou Hive, é mantida pelo próprio service de metadados. Alterações feitas no lado do Flink não afetam essas configurações.

Solução

• Para alterar a configuração de um catálogo registrado no console, exclua o catálogo e crie-o novamente com a nova configuração. Para mais informações, consulte o tópico de gerenciamento do catálogo correspondente, como Manage DLF-Legacy Catalog ou Manage Hive Catalog.

• Para modificar a configuração de um catálogo de terceiros, como um catálogo DLF, faça a alteração diretamente no console do service cloud correspondente.