Realtime Compute for Apache Flink:カタログに関するよくある質問

Hologres カタログジョブがリアルタイムデータを消費せずに終了する問題

事象

Flink ジョブが Hologres から既存データを読み取った後に完了して終了します。新しいレコードを待機しません。

原因

Realtime Compute for Apache Flink は、デフォルトで Hologres からストリーミングモードではなく、バッチモードでデータを読み取ります。

解決策

SQL クエリに /*+ OPTIONS('binlog'='true') */ ヒントを追加して、ストリーミングモードに切り替えてください。

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

CREATE TABLE AS での CatalogTableProvider インターフェイスエラー

エラーメッセージ

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.

原因

シンクテーブルのカタログが CatalogTableProvider インターフェイスを実装していないため、CREATE TABLE AS (CTAS) ステートメントで使用できません。たとえば、MongoDB カタログはシンクとして CTAS をサポートしていません。

解決策

CTAS をサポートするソーステーブルとシンクテーブルに切り替えてください。サポートされているストレージシステムの完全なリストについては、「CREATE TABLE AS (CTAS)」をご参照ください。

MySQL カタログクエリの低速応答またはサーバー例外

MySQL カタログのクエリで応答が遅い、またはサーバー例外が発生する場合、通常は Flink と MySQL 間のネットワークレイテンシーが高いか、接続が不安定であることが原因です。状況に応じた対処を行ってください。

-- Author:        izha****69041.onaliyun.com
-- Created Time:  2022-06-30 14:30:42
-- 説明：ここに説明を記述します

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

原因 1: パブリック帯域幅の不足

Realtime Compute for Apache Flink から MySQL へのパブリック帯域幅が狭すぎるかどうかを確認してください。デフォルトのパブリック帯域幅は 2 Mbit/s 程度と低い場合があります。また、クロスリージョンアクセスが発生しているかどうかも確認してください。帯域幅の確認方法については、「パブリック帯域幅を確認する方法」をご参照ください。

原因 2: ネットワーク接続の不安定

MySQL インスタンスの interactive_timeout および wait_timeout パラメーターの値を増やして、接続切断の頻度を減らしてください。

原因 3: 接続タイムアウトが短すぎる

MySQL カタログのテーブルを作成する際に、'connect.timeout'='120s' オプションを WITH 句に追加してください。

-- CREATE TABLE ステートメントの WITH 句に追加
'connect.timeout' = '120s'

E-MapReduce (EMR) メタサービスからの STS トークン取得失敗

エラーメッセージ

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

原因

hive-site.xml 設定ファイルに、dlf.catalog.akMode の設定として EMR_AUTO が含まれています。この設定は E-MapReduce (EMR) クラスター内でのみ機能します。EMR クラスター外で実行すると、Hive カタログの初期化が失敗します。

解決策

hive-conf-dir ディレクトリ内の hive-site.xml を更新してください。

1. 次のプロパティを削除してください。

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

2. アクセスキー ID とシークレットアクセスキーを追加してください。

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

コンソールまたは ALTER CATALOG によるカタログ設定変更の不反映

原因

• Realtime Compute for Apache Flink は、登録済みのカタログの設定変更をサポートしていません。コンソールにはそのような変更を行うための項目がなく、ALTER CATALOG ステートメントではカタログ設定を変更できません。

• DLF や Hive などのサードパーティカタログの設定は、独自のメタデータサービスによって管理されています。Flink 側で行った変更は、これらの設定に影響しません。

解決策

• コンソールに登録されたカタログの設定を変更するには、カタログを削除して、新しい設定で再作成してください。詳細については、「DLF-Legacy カタログの管理」や「Hive カタログの管理」など、対応するカタログの管理トピックをご参照ください。

• DLF カタログなどのサードパーティカタログ自体の設定を変更するには、対応するクラウドサービスのコンソールで変更してください。