Realtime Compute for Apache Flink:コネクタ

• Kafka

  • Kafka から JSON データを取得する方法

  • ネットワーク接続が正常にもかかわらず、Flink が Kafka からデータを消費または書き込みできない理由

  • イベント時間ウィンドウを使用した Kafka ソースからデータが出力されない

  • Kafka でのオフセットのコミットの目的

  • Kafka コネクタによるネストされた JSON データの解析

  • セキュリティが有効な Kafka クラスターへの接続

  • フィールド名の競合の解決

  • Kafka ソースからの予期しないレイテンシーへの対処

  • エラー： 'upsert-kafka' table require PRIMARY KEY

• DataHub

  • DataHub トピックのスケーリング後のジョブの回復

  • 消費済み DataHub トピックの削除

  • DataHub テーブル作成時の "NoPermissionException: dhs:ListShard" エラー

• MaxCompute

  • endPoint と tunnelEndpoint の説明

  • 全量および増分 MaxCompute ソースのデータ読み取り

  • MaxCompute ソースでの、ジョブ開始後に追加されたデータの読み取り

  • 再開された MaxCompute ソースジョブの同時実行数の変更

  • MaxCompute ソースが開始位置より前のパーティションを読み取る理由

  • 増分 MaxCompute ソースにおける不完全な新しいパーティションの処理

  • MaxCompute コネクタエラー： ErrorMessage=Authorization Failed [4019], You have NO privilege

  • 増分 MaxCompute ソースの startPartition パラメーターの設定

  • 増分 MaxCompute ソースの読み取り開始が遅延する問題

  • Partition パラメータの設定

  • MaxCompute ソースを使用したジョブの起動時のスタックまたは遅延

  • トンネルの選択

  • MaxCompute 結果テーブルでの重複データの処理

  • MaxCompute 結果テーブルでの実行時エラー： 'Invalid partition spec'

  • MaxCompute 結果テーブルでの実行時エラー： 'No more available blockId'

  • max_pt() と max_pt_with_done() の違い

  • SHUFFLE_HASH ヒントの使用

  • CacheReloadTimeBlackList パラメータの設定

• MySQL

  • エラー： java.io.EOFException: SSL peer shut down incorrectly

• ApsaraDB RDS for MySQL

  • データ同期中の bigint unsigned 主キーの型変更

  • Flink による RDS への書き込み：更新と挿入の動作

  • RDS テーブルのユニークインデックスに対する GROUP BY の使用

  • MySQL の INT UNSIGNED が Flink SQL で異なる型を必要とする理由

  • エラー： Incorrect string value: '\xF0\x9F\x98\x80\xF0\x9F...' for column 'test' at row 1

  • MySQL への書き込み時のデッドロック

  • MySQL スキーマ更新後にダウンストリームスキーマが変更されない

  • 'finish split response timeout' 例外の原因

  • MySQL CDC 全量同期中のスキーマ変更の影響

  • CTAS/CDAS 同期中のサポートされていないスキーマ変更によるジョブ失敗の解決

• ClickHouse

  • ClickHouse 結果テーブルでのデータの撤回と更新

  • ClickHouse 結果テーブルでのデータの可視性

• Print

  コンソールでの print 結果の表示

• Tablestore

  ディメンションテーブルへの JOIN でデータが返されない

• ApsaraMQ for RocketMQ

  • エラー： IllegalArgumentException: timeout value is negative

  • トピックのスケールアウト時に RocketMQ がパーティション変更を検出する方法

• Hologres

  • エラー： BackPressure Exceed reject Limit

  • エラー： remaining connection slots are reserved for non-replication superuser connections

  • エラー： no table is defined in publication

  • Hologres シンクのチェックポイント間隔とデータの可視性の関係

  • デプロイ時の 'permission denied for database' 例外

  • チェックポイントからの復元時の 'table id ...' 例外

  • JDBC モードでの binlog データ消費におけるデータ精度の問題

  • 同じ名前のテーブルを削除して再作成した後のエラー

  • Hologres からの読み取り時の ClassCastException

• Log Service

  • エラー： Caused by: com.aliyun.openservices.aliyun.log.producer.errors.LogSizeTooLargeException: the logs is 8785684 bytes which is larger than MAX_BATCH_SIZE_IN_BYTES 8388608

  • 失敗したジョブの復元時における TaskManager の OOM エラー

• Paimon

  • Paimon ソースの起動位置の設定

  • パーティションの自動有効期限の設定

  • 'Heartbeat of TaskManager timed out' による Paimon 書き込みジョブの失敗

  • 'Sink materializer must not be used with Paimon sink' エラーによる Paimon 書き込みジョブの失敗

  • Paimon への書き込み時の競合エラー

  • 'File xxx not found' エラーによる Paimon 読み取りジョブの失敗

  • 'No space left on device' エラーによる Paimon ジョブの失敗

  • OSS 上の多数の Paimon ファイルの管理

  • Paimon データの可視性とチェックポイント間隔

  • Paimon への書き込みジョブでのメモリの緩やかな増加

• Hudi

  • ストレージにデータが見つからない

  • Hudi での重複データの処理

  • マージオンリードテーブルにログファイルのみが存在する理由

• AnalyticDB for MySQL (ADB)

  エラー： multi-statement be found

• カスタムコネクタ

  • エラー： No suitable driver found for

• Elasticsearch

  • Flink による Elasticsearch への書き込みにおけるデータ損失または上書き

Flink を使用した Kafka からの JSON データの取得

• 標準的な JSON データを取得するには、「JSON Format」をご参照ください。

• ネストされた JSON データを取得するには、ソーステーブルの DDL で JSON オブジェクトをROW 型として定義します。シンクテーブルの DDL で、取得するキーを宣言します。次に、DML ステートメントを使用してキーにアクセスし、その値を抽出します。以下にコード例を示します。

  • サンプルデータ

    {
        "a":"abc",
        "b":1,
        "c":{
            "e":["1","2","3","4"],
            "f":{"m":"567"}
        }
    }

  • ソーステーブル DDL

    CREATE TEMPORARY TABLE `kafka_table` (
      `a` VARCHAR,
       b int,
      `c` ROW<e ARRAY<VARCHAR>,f ROW<m VARCHAR>>  -- 'c' は、Flink の ROW 型にマッピングされる JSON オブジェクトです。'e' は、ARRAY 型にマッピングされる JSON 配列です。
    ) WITH (
      'connector' = 'kafka',
      'topic' = 'xxx',
      'properties.bootstrap.servers' = 'xxx',
      'properties.group.id' = 'xxx',
      'format' = 'json',
      'scan.startup.mode' = 'xxx'
    );

  • シンクテーブル DDL

    CREATE TEMPORARY TABLE `sink` (
     `a` VARCHAR,
      b INT,
      e VARCHAR,
      `m` varchar
    ) WITH (
      'connector' = 'print',
      'logger' = 'true'
    );

  • DML ステートメント

    INSERT INTO `sink`
      SELECT 
      `a`,
      b,
      c.e[1], -- Flink は配列に対して 1 ベースのインデックスを使用します。この例では、インデックス [1] を使用して配列から最初の要素を抽出します。
      c.f.m
    FROM `kafka_table`;

  • 結果

    409  2021-04-08 10:13:11,214 INFO  org.apache.flink.kafka.shaded.org.apache.kafka.common.utils.AppInfoParser [] -
    410  2021-04-08 10:13:11,215 INFO  org.apache.flink.kafka.shaded.org.apache.kafka.common.utils.AppInfoParser [] - Kafka commitId: cb8625948210849f
    411  2021-04-08 10:13:11,215 INFO  org.apache.flink.kafka.shaded.org.apache.kafka.common.utils.AppInfoParser [] - Kafka startTimeMs: 1617847991214
    412  2021-04-08 10:13:11,270 INFO  org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.KafkaConsumer [] - [Consumer clientId=consumer-lb_test-2, groupId=lb_test] Subscribed to partition(s): lb_test-0, lb_test-1, lb_test-2, lb_test-3, lb_test-4, lb_test-5
    413  2021-04-08 10:13:11,280 INFO  org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.KafkaConsumer [] - [Consumer clientId=consumer-lb_test-2, groupId=lb_test] Seeking to offset 1 for partition lb_test-0
    414  2021-04-08 10:13:11,285 INFO  org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.KafkaConsumer [] - [Consumer clientId=consumer-lb_test-2, groupId=lb_test] Seeking to offset 0 for partition lb_test-1
    415  2021-04-08 10:13:11,285 INFO  org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.KafkaConsumer [] - [Consumer clientId=consumer-lb_test-2, groupId=lb_test] Seeking to offset 0 for partition lb_test-2
    416  2021-04-08 10:13:11,285 INFO  org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.KafkaConsumer [] - [Consumer clientId=consumer-lb_test-2, groupId=lb_test] Seeking to offset 0 for partition lb_test-3
    417  2021-04-08 10:13:11,290 INFO  org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.KafkaConsumer [] - [Consumer clientId=consumer-lb_test-2, groupId=lb_test] Seeking to offset 0 for partition lb_test-4
    418  2021-04-08 10:13:11,291 INFO  org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.KafkaConsumer [] - [Consumer clientId=consumer-lb_test-2, groupId=lb_test] Seeking to offset 0 for partition lb_test-5
    419  2021-04-08 10:13:11,302 INFO  org.apache.flink.kafka.shaded.org.apache.kafka.clients.Metadata [] - [Consumer clientId=consumer-lb_test-2, groupId=lb_test] Cluster ID: -1flJPwnTvuGFSuyCtU1hw
    420  2021-04-08 10:15:31,597 INFO  org.apache.flink.api.common.functions.util.PrintSinkOutputWriter [] - +I(abc,1,1,567)

Flink が Kafka のデータを消費または書き込みできない

• 原因

  Flink と Kafka の間にプロキシやポートマッピングなどの転送メカニズムが存在する場合、Kafka クライアントはプロキシのアドレスではなく、Kafka サーバーの内部ネットワークアドレスを取得します。その結果、ネットワークパスが確立されていても、Flink は Kafka クラスターに接続できますが、データの消費または書き込みができません。

  Flink Kafka コネクターと Kafka サーバー間の接続プロセスは、2 つのステップで構成されます：

  1. Kafka クライアントは Kafka ブローカーからメタデータを取得します。このメタデータには、クラスター内のすべてのブローカーのネットワークアドレスが含まれています。

  2. その後、Flink コネクターはこれらのネットワークアドレスを使用して、データを消費または書き込みます。

• トラブルシューティング

  以下の手順に従って、Flink と Kafka の間にプロキシやポートマッピングなどの転送メカニズムが存在するかどうかを判断してください：

  1. ZooKeeper コマンドラインツール (zkCli.sh または zookeeper-shell.sh) を使用して、Kafka クラスターが使用する ZooKeeper クラスターにログインします。

  2. クラスターに適したコマンドを実行して、Kafka ブローカーのメタデータを取得します。

     通常、get /brokers/ids/0 コマンドを使用して Kafka ブローカーのメタデータを取得できます。接続アドレスは endpoints フィールドにあります。たとえば、ZooKeeper シェルを使用して接続し、get /brokers/ids/0 を実行してブローカーの登録情報を表示します。返された JSON の endpoints フィールドに設定されているアドレスに注意してください：

     # bin/zookeeper-shell.sh localhost:2181
     Connecting to localhost:2181
     Welcome to ZooKeeper!
     JLine support is disabled
     WATCHER::
     WatchedEvent state:SyncConnected type:None path:null
     get /brokers/ids/0
     {"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"},"endpoints":["PLAINTEXT://116.62.xxx:9092"],"jmx_port":-1,"host":"116.62.xxx","timestamp":"1614840078030","port":9092,"version":4}

  3. ping や telnet などのコマンドを使用して、Flink 環境から endpoints フィールドのアドレスへの接続性をテストしてください。

     接続に失敗した場合、Flink と Kafka の間にプロキシやポートマッピングなどの転送メカニズムが存在することを示します。

• ソリューション

  • 転送メカニズムを使用しないでください。代わりに、Flink と Kafka の間に直接のネットワークパスを確立してください。これにより、Flink は Kafka のメタデータに記載されているエンドポイントに直接接続できます。

  • Kafka 管理者に連絡して、Kafka ブローカーの advertised.listeners プロパティに転送アドレスを設定してもらってください。これにより、Kafka クライアントは正しい転送アドレスを含むメタデータを取得できます。

    説明

    Kafka ブローカーのリスナーへのプロキシアドレスの追加は、Kafka バージョン 0.10.2.0 以降でのみサポートされています。

  この仕組みの詳細については、「KIP-103: Separate Internal and External traffic」と「Kafka client cannot connect to brokers」をご参照ください。

Flink と Kafka 間のネットワーク接続が確認されても問題が解決しない場合は、以下のネットワーク以外の原因を確認してください：

確認 1： 開始オフセット戦略

Kafka ソーステーブルの DDL の WITH 句にある scan.startup.mode パラメーターを確認してください。値が latest-offset の場合、Flink はジョブの開始後に書き込まれたメッセージのみを読み取ります。ジョブの開始後に新しいメッセージが届かない場合、ジョブはデータを消費していないように見えます。

ジョブの開始後に新しいデータが生成されていることを確認するには、Kafka コンシューマークライアントを使用してトピックをリアルタイムで監視してください。

確認 2： データフォーマットの不一致

Kafka ソーステーブルの WITH 句にある format パラメーターが、Kafka トピック内のメッセージの実際のエンコーディングと一致していることを確認してください。フォーマットの不一致はデシリアライズの失敗を引き起こすため、ジョブがメッセージをサイレントにスキップしたり、出力が生成されなくなったりする可能性があります。

実際のメッセージフォーマットを確認するには、Kafka コンシューマークライアントを使用してトピックから RAW バイトを読み取り、ペイロードの構造を調べてください。

Kafka のイベント時間タイムウィンドウでデータが出力されない

• 問題

  Kafka ソーステーブルでイベント時間タイムウィンドウを使用すると、ジョブから出力されません。

• 原因

  アイドル状態の Kafka パーティションによってウォーターマークが進まなくなり、イベント時間タイムウィンドウからの出力が停止する可能性があります。

• ソリューション

  1. すべてのパーティションがデータを受信することを確認してください。

  2. ソースのアイドル検出を有効にするには、次のコードを [その他の設定] セクションに追加して変更を保存します。詳細な手順については、「カスタムジョブランタイムパラメーターの設定方法」をご参照ください。

     table.exec.source.idle-timeout: 5

     table.exec.source.idle-timeout パラメーターの詳細については、「設定」をご参照ください。

Kafka でのオフセットのコミット

Kafka でのオフセットのコミットは、処理済みデータの位置を追跡してデータの重複やデータ損失を防ぎ、ストリーム処理における一貫性と信頼性を確保します。チェックポイントが正常に完了すると、Flink は対応する読み取りオフセットを Kafka にコミットします。チェックポイントが有効になっていない場合、またはチェックポイント間隔が長すぎる場合、Kafka にコミットされたオフセットは古くなり、データ再処理やデータ損失につながる可能性があります。

Kafka コネクタによるネストされた JSON の解析

{"data":[{"cola":"test1","colb":"test2"},{"cola":"test1","colb":"test2"},{"cola":"test1","colb":"test2"},{"cola":"test1","colb":"test2"},{"cola":"test1","colb":"test2"}]}

セキュリティが有効な Kafka クラスターへの接続

1. Kafka テーブル DDL の WITH 句に、認証と暗号化のセキュリティ設定を追加します。オプションの完全なリストについては、「SECURITY」をご参照ください。

   重要

   すべてのセキュリティ設定パラメーターの先頭に properties. を付けてください。

   • 次の例では、PLAIN SASL メカニズムを使用し、JAAS 設定を提供するように Kafka テーブルを設定する方法を示します。

     CREATE TABLE KafkaTable (
       `user_id` BIGINT,
       `item_id` BIGINT,
       `behavior` STRING,
       `ts` TIMESTAMP(3) METADATA FROM 'timestamp'
     ) WITH (
       'connector' = 'kafka',
       ...
       'properties.security.protocol' = 'SASL_PLAINTEXT',
       'properties.sasl.mechanism' = 'PLAIN',
       'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule required username=\"username\" password=\"password\";'
     );

   • 次の例では、SASL_SSL セキュリティプロトコルと SCRAM-SHA-256 SASL メカニズムを使用する方法を示します。

     CREATE TABLE KafkaTable (
       `user_id` BIGINT,
       `item_id` BIGINT,
       `behavior` STRING,
       `ts` TIMESTAMP(3) METADATA FROM 'timestamp'
     ) WITH (
       'connector' = 'kafka',
       ...
       'properties.security.protocol' = 'SASL_SSL',
       /* SSL 設定 */
       /* サーバーから提供されたトラストストア (CA 証明書) へのパス */
       'properties.ssl.truststore.location' = '/flink/usrlib/kafka.client.truststore.jks',
       'properties.ssl.truststore.password' = 'test1234',
       /* クライアント側の認証が必要な場合は、キーストア (秘密鍵) へのパスを設定します */
       'properties.ssl.keystore.location' = '/flink/usrlib/kafka.client.keystore.jks',
       'properties.ssl.keystore.password' = 'test1234',
       /* SASL 設定 */
       /* SASL メカニズムを SCRAM-SHA-256 として設定します */
       'properties.sasl.mechanism' = 'SCRAM-SHA-256',
       /* JAAS を設定します */
       'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.scram.ScramLoginModule required username=\"username\" password=\"password\";'
     );

     説明

     • properties.sasl.mechanism が SCRAM-SHA-256 の場合、properties.sasl.jaas.config には org.apache.flink.kafka.shaded.org.apache.kafka.common.security.scram.ScramLoginModule を使用してください。

     • properties.sasl.mechanism が PLAIN の場合、properties.sasl.jaas.config には org.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule を使用してください。

2. ジョブの [追加の依存ファイル] セクションで、証明書、公開鍵、秘密鍵など、必要なすべてのファイルをアップロードします。

   プラットフォームは、アップロードされたファイルを /flink/usrlib ディレクトリに保存します。アップロード手順については、「ジョブのデプロイメント」をご参照ください。

   重要

   Kafka ブローカーの認証メカニズムが SASL_SSL で、クライアント側のメカニズムが SASL_PLAINTEXT の場合、検証中にジョブが OutOfMemory 例外で失敗します。この問題を解決するには、クライアント側とサーバー側の認証メカニズムが一致していることを確認してください。

フィールド名の競合

• 問題

  Kafka データソースは、メッセージをキー用と値用の 2 つの独立した JSON 文字列にシリアル化します。このシナリオでは、キーと値の両方に、以下の例の id フィールドのように、同じ名前のフィールドが含まれています。このデータを Flink テーブルに直接解析すると、フィールド名の競合が発生します。

  • キー

    {
       "id": 1
    }

  • 値

    {
       "id": 100,
       "name": "flink"
    }

• ソリューション

  この問題を回避するには、key.fields-prefix プロパティを使用します。

  CREATE TABLE kafka_table (
    -- キーと値のフィールド用のカラムを定義します
    key_id INT,
    id INT,
    name STRING
  ) WITH (
    'connector' = 'kafka',
    'topic' = 'test_topic',
    'properties.bootstrap.servers' = 'localhost:9092',
    'format' = 'json',
    'json.ignore-parse-errors' = 'true',
    -- キーのフィールドとデータ型を指定します
    'key.format' = 'json',
    'key.fields' = 'id',
    'value.format' = 'json',
    'value.fields' = 'id, name',
    -- キーのフィールドにプレフィックスを追加します
    'key.fields-prefix' = 'key_'
  );

  key.fields-prefix プロパティを key_ に設定すると、コネクタはメッセージ キーのすべてのフィールドに key_ プレフィックスを追加します。例えば、キーの id フィールドは Flink テーブルの key_id カラムになります。これにより、id カラムにマッピングされる値の id フィールドとの競合を防ぎます。

  SELECT * FROM kafka_table; クエリを実行すると、以下の出力が返されます。

  key_id: 1,
  id: 100,
  name: flink

Kafka ソースからの高いレイテンシーのトラブルシューティング

• 問題

  Kafka ソーステーブルから読み取りを行うと、currentEmitEventTimeLag メトリクスが 50 年を超える値を示します。例えば、複数の Realtime Compute for Apache Flink の SQL ジョブが [running] 状態にありますが、[business latency] カラムには、19160d 1h 59m 28s のように 19,160 日を超える異常に高い値が表示されます。

• トラブルシューティング

  1. まず、ジョブが JAR ジョブか SQL ジョブかを判断します。

     JAR ジョブの場合、pom.xml ファイルが Realtime Compute for Apache Flink 提供の Kafka 依存関係を使用していることを確認してください。オープンソースバージョンのコネクタは、これらのメトリクスを報告しません。

  2. アップストリームの Kafka トピックのすべてのパーティションがリアルタイムでデータを受信しているかどうかを確認します。

  3. Kafka メッセージのメタデータの タイムスタンプ が 0 または null であるかどうかを確認します。

     Kafka ソースのレイテンシーは、現在の時刻からメッセージのタイムスタンプを引くことで計算されます。メッセージにタイムスタンプがない場合、レイテンシーは 50 年以上と表示されることがあります。タイムスタンプは、以下のいずれかの方法で確認できます：

     • SQL ジョブの場合、メタデータカラムを定義することでメッセージのタイムスタンプを取得できます。詳細については、「Kafka ソーステーブル」をご参照ください。

       CREATE TEMPORARY TABLE sk_flink_src_user_praise_rt (
           `timestamp` BIGINT,
           `kafka_timestamp` TIMESTAMP METADATA,  -- メタデータのタイムスタンプ。
           ts as to_timestamp (
             from_unixtime (`timestamp`, 'yyyy-MM-dd HH:mm:ss')
           ),
           watermark for ts as ts - interval '5' second
         ) WITH (
           'connector' = 'kafka',
           'topic' = '',
           'properties.bootstrap.servers' = '',
           'properties.group.id' = '',
           'format' = 'json',
           'scan.startup.mode' = 'latest-offset',
           'json.fail-on-missing-field' = 'false',
           'json.ignore-parse-errors' = 'true'
         );

     • KafkaConsumer クライアントを使用してメッセージを読み取り、そのタイムスタンプを検査する簡単な Java プログラムを作成します。

エラー: 'upsert-kafka' テーブルには主キーが必要です

• 問題

  ) WITH (
      'connector' = 'upsert-kafka',
      'topic' = 'flow_stay_duration',
      'properties.bootstrap.servers' = 'xxx',
      'key.format' = 'avro',
      'value.format' = 'avro'
  );
      insert into sink_ad_data_device_info
  org.apache.flink.table.api.ValidationException: SQL validation failed. Unable to create a sink for writing table 'vvp.default.sink_ad_data_device_info'.
  The cause is following: 'upsert-kafka' tables require to define a PRIMARY KEY constraint. The PRIMARY KEY specifies which columns should be read from or write to the Kafka message key. The PRIMARY KEY also defines records in the 'upsert-kafka' table should update or delete on which keys.
  Table options are:
  'connector'='upsert-kafka'
  'key.format'='avro'
  'properties.bootstrap.servers'='xxx'
  'topic'='flow_stay_duration'
  'value.format'='avro'
      at org.apache.flink.table.sqlserver.utils.FormatValidatorExceptionUtils.newValidationException(FormatValidatorExceptionUtils.java:41)
      at org.apache.flink.table.sqlserver.utils.ErrorConverter.formatException(ErrorConverter.java:123)
      at org.apache.flink.table.sqlserver.utils.ErrorConverter.toErrorDetail(ErrorConverter.java:60)
      at org.apache.flink.table.sqlserver.utils.ErrorConverter.toGrpcException(ErrorConverter.java:54)
      at org.apache.flink.table.sqlserver.FlinkSqlServiceImpl.validateAndGeneratePlan(FlinkSqlServiceImpl.java:979)
      at org.apache.flink.table.sqlserver.proto.FlinkSqlServiceGrpc$MethodHandlers.invoke(FlinkSqlServiceGrpc.java:3283)
      at io.grpc.stub.ServerCalls$UnaryServerCallHandler$UnaryServerCallListener.onHalfClose(ServerCalls.java:172)
      at io.grpc.internal.ServerCallImpl$ServerStreamListenerImpl.halfClosed(ServerCallImpl.java:331)

• 原因

  このエラーは、DDL に PRIMARY KEY がないために発生します。シンクテーブルとして使用する場合、upsert-kafka コネクタはアップストリームロジックからの変更ログストリームを消費します。このコネクタは、INSERT および UPDATE_AFTER データを Kafka に書き込みます。DELETE 操作の場合、対応するキーのメッセージが削除されたことを示すために、null 値を持つメッセージを書き込みます。Flink は、PRIMARY KEY カラムを使用してデータをパーティション分割します。これにより、同じキーを持つメッセージが順序付けられ、対応する更新または削除メッセージが同じパーティションに格納されることが保証されます。

• 解決策

  DDL で PRIMARY KEY を定義してください。

トピックの分割・スケールイン後の Flink ジョブの回復

Flink ジョブが読み取っている DataHub トピックを分割またはスケールインすると、ジョブは失敗ループに入り、自動的に回復できなくなります。この問題を解決するには、ジョブを再起動してください。

アクティブなコンシューマーを持つトピックの削除

アクティブなコンシューマーを持つ DataHub トピックは、削除または再作成できません。

endPoint および tunnelEndpoint パラメーター

DataHub テーブルの作成が NoPermissionException: dhs:ListShard で失敗する

• 症状

  Flink ジョブが DataHub のソーステーブルまたはシンクテーブルをデプロイすると、次のようなエラーで失敗します。

  NoPermissionException: You have no permission to perform this action. Action: dhs:ListShard

• 原因

  このエラーは、DataHub DDL の非標準の WITH パラメーター名が原因です。DataHub コネクタでは、認証情報を accessId および accessKey として指定する必要があります。代わりにドット形式の access.id および access.key を使用すると、コネクタは認証情報フィールドを認識できず、認証できません。その結果、コネクタは有効な認証情報なしでシャードをリストしようとし、DataHub は NoPermissionException を返します。

  エラーメッセージには、dhs:ListShard 権限が不足していると表示されますが、根本原因は、実際の IAM 権限の不足ではなく、認証情報パラメーターが認識されないことです。

• 解決策

  DataHub DDL で、access.id を accessId に、access.key を accessKey に変更します。既存のテーブルを DROP し、修正された WITH 句で再作成します。

  不正な設定：

  CREATE TABLE datahub_source (...) WITH (
    'connector'       = 'datahub',
    'endPoint'        = 'https://dh-cn-hangzhou.aliyuncs.com',
    'project'         = 'your_project',
    'topic'           = 'your_topic',
    'access.id'       = 'your-access-key-id',      -- 不正：ドット形式は認識されません
    'access.key'      = 'your-access-key-secret'   -- 不正：ドット形式は認識されません
  );

  正しい設定：

  CREATE TABLE datahub_source (...) WITH (
    'connector'  = 'datahub',
    'endPoint'   = 'https://dh-cn-hangzhou.aliyuncs.com',
    'project'    = 'your_project',
    'topic'      = 'your_topic',
    'accessId'   = 'your-access-key-id',      -- 正しい
    'accessKey'  = 'your-access-key-secret'   -- 正しい
  );

  サポートされている WITH パラメーターの完全なリストについては、DataHub コネクタドキュメントをご参照ください。

MaxCompute ソースによる完全読み取りと増分読み取り

MaxCompute ソースは、MaxCompute トンネルを通じて完全読み取りと増分読み取りを行います。読み取りスループットは、MaxCompute トンネルの帯域幅によって制限されます。

MaxCompute ソーステーブルで追記データを読み取れますか。

いいえ。Flink ジョブの開始後、ソーステーブルまたはパーティションに追記された新しいデータは読み取りません。これは、ソースが読み取り中の場合も、読み取りが完了している場合も同様です。この方法でデータを追記すると、ジョブのフェイルオーバーが発生する場合もあります。

一時停止した MaxCompute ジョブの同時実行数の変更

useNewApi オプションが有効な MaxCompute ソーステーブル (デフォルトで有効) では、ストリーミングモードのジョブは、一時停止して再開した後に同時実行数を変更できます。MaxCompute ソーステーブルは、一致するパーティションを順次読み取ります。パーティションを読み取る際、そのパーティション内のデータを並列オペレーター間で分散します。同時実行数を変更しても、一時停止前に処理されていたパーティションのデータ分散には影響しません。新しい並列度は、ジョブが次のパーティションの処理を開始する際にのみ有効になります。その結果、ジョブが単一の大きなパーティションを処理している場合、同時実行数を増やしてジョブを再開すると、一部の並列オペレーターのみがデータを読み取る可能性があります。

バッチジョブ、または useNewApi オプションが false に設定されているジョブでは、同時実行数を変更できません。

開始位置が 2019-10-11 00:00:00 の場合に MaxCompute が過去のパーティションを読み取る現象

新しいパーティションにおける不完全なデータ読み取りの防止

MaxCompute コネクタの認可エラー

• エラーの詳細

  ジョブの実行中に、フェールオーバーページまたは TaskManager.log ファイルに次のエラーが表示されます。

  ErrorMessage=Authorization Failed [4019], You have NO privilege'ODPS:***'

• 原因

  MaxCompute DDL 定義で指定されたユーザーアイデンティティ情報に、MaxCompute へのアクセスに必要な権限が付与されていません。

• 解決策

  Alibaba Cloud アカウント、RAM ユーザー、または RAM ロールを使用して認証します。詳細については、「ユーザー認証」をご参照ください。

startPartition パラメーターの設定

インクリメンタル MaxCompute ジョブの起動遅延

partition パラメーターの設定

パーティションからの読み取り

• 静的パーティションからの読み取り

  ソーステーブルまたはディメンションテーブルの静的パーティションから読み取る場合は、partition パラメーターを次のように設定します。

  手順	説明	例
  1	ディメンションテーブルの場合、各パーティションを 'partition_column_name=partition_value' として指定します。パーティション値は固定値である必要があります。 ソーステーブルの場合、各パーティションを 'partition_column_name=partition_value' として指定します。パーティション値は、固定値またはワイルドカード (*) を含む値にすることができます。ワイルドカードは、空文字列を含む任意の文字列に一致します。	パーティション列 dt の値が 20220901 のデータを読み取るには、dt=20220901 を指定します。 dt 列の値が 202209 で始まるパーティションからデータを読み取るには、dt=202209*と指定します (ソーステーブルにのみ適用されます)。 dt 列のパーティションのうち、値が 2022 で始まり 01 で終わるものからデータを読み取るには、dt=2022*01 を指定します (ソーステーブルにのみ適用されます)。 dt 列のすべてのパーティションからデータを読み取るには、dt=* を指定します (ソーステーブル にのみ適用されます)。
  2	手順 1 のパーティション文字列をパーティションレベルで昇順にソートし、スペースを入れずにカンマで結合します。結果の文字列が partition パラメーターの値になります。 最初のいくつかのパーティションレベルのみを指定することもできます。	テーブルには、dt という単一の第 1 レベルのパーティションがあります。dt=20220901 パーティションからデータを読み取るには、'partition' = 'dt=20220901' を指定します。 テーブルには、第 1 レベルのパーティション dt、第 2 レベルのパーティション hh、および第 3 レベルのパーティション mm の 3 つのパーティションレベルがあります。dt=20220901、hh=08、および mm=10 のデータを読み取るには、'partition' = 'dt=20220901,hh=08,mm=10' を指定します。 同じテーブルで、dt=20220901、hh=08、および mm の任意の値からデータを読み取るには、'partition' = 'dt=20220901,hh=08' or 'partition' = 'dt=20220901,hh=08,mm=*' と指定します。 同じテーブルで、dt=20220901、hh が任意、mm=10 のデータを読み取るには、'partition' = 'dt=20220901,hh=*,mm=10' と指定します。

  これらの手順でパーティションフィルタリングの要件が満たされない場合は、SQL ステートメントの WHERE 句にフィルター条件を追加できます。これにより、SQL オプティマイザによるパーティションプッシュダウンを使用したフィルタリングが可能になります。例えば、2 つのパーティションレベル (dt と hh) を持つテーブルから、dt が '20220901' と '20220903' の間、hh が '09' と '17' の間にあるパーティションを読み取るには、次のような SQL ステートメントを使用します。

  CREATE TABLE maxcompute_table (
    content VARCHAR,
    dt VARCHAR,
    hh VARCHAR
  ) PARTITIONED BY (dt, hh) WITH ( 
     -- パフォーマンス向上のため、PARTITIONED BY でパーティション列を指定し、SQL オプティマイザによるパーティションプッシュダウンを有効化します。
    'connector' = 'odps',
    ... -- accessId などの必須パラメーターを入力します。SQL オプティマイザにパーティションをフィルタリングさせるには、「partition」パラメーターを省略します。
  );
  SELECT content, dt, hh FROM maxcompute_table
  WHERE dt >= '20220901' AND dt <= '20220903' AND hh >= '09' AND hh <= '17'; -- WHERE 句でフィルター条件を指定します。

• 辞書順で最大のパーティションの読み取り

  • ソーステーブルまたはディメンションテーブルから辞書順で最大のパーティションを読み取るには、partition パラメーターを 'max_pt()' に設定します。

  • ソーステーブルまたはディメンションテーブルから辞書順で大きい方から 2 つのパーティションを読み取るには、partition パラメーターを 'max_two_pt()' に設定します。

  • ソーステーブルまたはディメンションテーブルから、対応する .done パーティションも持つ辞書順で最大のパーティションを読み取るには、partition パラメーターを 'max_pt_with_done()' に設定します。

  通常、辞書順で最大のパーティションは、最も新しく作成されたパーティションです。max_pt_with_done() オプションは、最新のパーティションのデータが準備できていない可能性があり、ディメンションテーブルに一時的に少し古いが完全なパーティションから読み取らせたい場合に便利です。

  パーティションのデータの準備ができたら、対応する空のパーティションも作成する必要があります。その名前は、データパーティション名に .done を付加したものです。例えば、dt=20220901 パーティションのデータの準備ができた後、dt=20220901.done という名前の空のパーティションを作成します。partition パラメーターを max_pt_with_done() に設定すると、ディメンションテーブルは対応する .done パーティションを持つパーティションからのみ読み取ります。.done パーティションのないデータパーティションは一時的に無視されます。詳細については、「max_pt() と max_pt_with_done() の違い」をご参照ください。

  説明

  ソーステーブルは、ジョブの開始時にのみ辞書順で最大のパーティションを決定します。すべてのデータを読み取った後に停止し、新しいパーティションは監視しません。新しいパーティションを継続的に読み取る必要がある場合は、増分ソーステーブルモードを使用します。ディメンションテーブルは、リフレッシュされるたびに最新のデータをチェックして読み取ります。

パーティションへの書き込み

• 静的パーティションへの書き込み

  結果テーブルの静的パーティションにデータを書き込むには、静的パーティションからの読み取りと同じ方法で partition パラメーターを設定できます。

  重要

  結果テーブルの partition パラメーターは、ワイルドカード (*) をサポートしていません。

• 動的パーティションへの書き込み

  データからパーティション値が導出される動的パーティションに書き込むには、partition パラメーターを、パーティションレベルの昇順でソートされたパーティション列名のコンマ区切りリストに設定します。たとえば、テーブルに dt、hh、mm の 3 つのパーティションレベルがある場合、'partition' = 'dt,hh,mm' と指定します。

MaxCompute ソーステーブルのジョブ起動の遅延

データチャネルの選択

MaxCompute 結果テーブルでのデータ重複

Flink ジョブによって書き込まれる MaxCompute 結果テーブルのデータ重複は、以下の原因で発生する可能性があります：

• ジョブのロジックを確認してください。MaxCompute 結果テーブルでプライマリキー制約が宣言されていても、Flink は外部ストレージへの書き込み時に一意性チェックを実行しません。さらに、MaxCompute の非トランザクションテーブルはプライマリキー制約をサポートしていません。したがって、Flink ジョブのロジックが重複データを生成する場合、これらの重複データは MaxCompute テーブルに書き込まれます。

• 複数の Flink ジョブが同じ MaxCompute テーブルに同時に書き込んでいるかどうかを確認してください。前述のように、MaxCompute はプライマリキー制約を強制しません。複数の Flink ジョブが同じ結果を生成する場合、テーブル内に重複レコードが作成されます。

• バッチトンネルの使用中に Flink ジョブがチェックポイントで失敗します。チェックポイント中に障害が発生すると、結果テーブルのデータが既にサーバーにコミットされている可能性があります。その結果、ジョブが最後に成功したチェックポイントから回復すると、最後に成功したチェックポイントと障害発生の間のデータを重複して書き込む可能性があります。

• ストリームトンネルの使用中に Flink ジョブのフェールオーバーが発生します。ストリームトンネルを使用して MaxCompute に書き込む場合、データはチェックポイント間で MaxCompute サーバーにコミットされます。ジョブがフェールオーバーして最新のチェックポイントから回復すると、チェックポイント完了後、フェールオーバー発生前に処理されたデータを重複して書き込む可能性があります。詳細については、「データチャネルの選択」をご参照ください。この種のデータ重複を防ぐには、バッチトンネルモードに切り替えることができます。

• バッチトンネルを使用する Flink ジョブがフェールオーバーするか、またはキャンセル後に再起動される場合 (例： Autopilot によってトリガーされた場合)、vvr-6.0.7-flink-1.15 より前のバージョンでは、ジョブはシャットダウン時に MaxCompute 結果テーブルにデータをコミットします。その結果、Flink ジョブが停止し、最後のチェックポイントから回復すると、最後のチェックポイントとシャットダウンの間に重複データを生成する可能性があります。この問題を解決するには、Flink のバージョンを vvr-6.0.7-flink-1.15 以降にアップグレードしてください。

MaxCompute ジョブの「無効なパーティション仕様」による失敗

• 原因： このエラーは、MaxCompute に書き込まれるデータのパーティション列に無効な値が含まれている場合に発生します。無効な値には、空の文字列、null 値、または等号 (=)、カンマ (,)、スラッシュ (/) を含む値が含まれます。

• 解決策： ソースデータのパーティション列の値が有効であることを確認してください。

MaxCompute ジョブにおける「No more available ブロック ID」エラー

• 原因：MaxCompute 結果テーブルに書き込まれたブロック数が上限を超えています。通常、少量のデータを頻繁にフラッシュしすぎることが原因です。

• 解決策： batchSize および flushIntervalMs パラメーターを調整してください。

SHUFFLE_HASH ヒントの使用

-- ソーステーブルと 3 つのディメンションテーブルを作成します。
CREATE TABLE source_table (k VARCHAR, v VARCHAR) WITH ( ... );
CREATE TABLE dim_1 (k VARCHAR, v VARCHAR) WITH ('connector' = 'odps', 'cache' = 'ALL', ... );
CREATE TABLE dim_2 (k VARCHAR, v VARCHAR) WITH ('connector' = 'odps', 'cache' = 'ALL', ... );
CREATE TABLE dim_3 (k VARCHAR, v VARCHAR) WITH ('connector' = 'odps', 'cache' = 'ALL', ... );
-- SHUFFLE_HASH ヒントに、分散するディメンションテーブルの名前を指定します。
SELECT /*+ SHUFFLE_HASH(dim_1), SHUFFLE_HASH(dim_3) */
k, s.v, d1.v, d2.v, d3.v
FROM source_table AS s
INNER JOIN dim_1 FOR SYSTEM_TIME AS OF PROCTIME() AS d1 ON s.k = d1.k
LEFT JOIN dim_2 FOR SYSTEM_TIME AS OF PROCTIME() AS d2 ON s.k = d2.k
LEFT JOIN dim_3 FOR SYSTEM_TIME AS OF PROCTIME() AS d3 ON s.k = d3.k;

CacheReloadTimeBlackList の設定

エラー: java.io.EOFException: SSL peer shut down incorrectly

• エラーの詳細

  Caused by: java.io.EOFException: SSL peer shut down incorrectly
      at sun.security.ssl.SSLSocketInputRecord.decodeInputRecord(SSLSocketInputRecord.java:239) ~[?:1.8.0_302]
      at sun.security.ssl.SSLSocketInputRecord.decode(SSLSocketInputRecord.java:190) ~[?:1.8.0_302]
      at sun.security.ssl.SSLTransport.decode(SSLTransport.java:109) ~[?:1.8.0_302]
      at sun.security.ssl.SSLSocketImpl.decode(SSLSocketImpl.java:1392) ~[?:1.8.0_302]
      at sun.security.ssl.SSLSocketImpl.readHandshakeRecord(SSLSocketImpl.java:1300) ~[?:1.8.0_302]
      at sun.security.ssl.SSLSocketImpl.startHandshake(SSLSocketImpl.java:435) ~[?:1.8.0_302]
      at com.mysql.cj.protocol.ExportControlled.performTlsHandshake(ExportControlled.java:347) ~[?:?]
      at com.mysql.cj.protocol.StandardSocketFactory.performTlsHandshake(StandardSocketFactory.java:194) ~[?:?]
      at com.mysql.cj.protocol.a.NativeSocketConnection.performTlsHandshake(NativeSocketConnection.java:101) ~[?:?]
      at com.mysql.cj.protocol.a.NativeProtocol.negotiateSSLConnection(NativeProtocol.java:308) ~[?:?]
      at com.mysql.cj.protocol.a.NativeAuthenticationProvider.connect(NativeAuthenticationProvider.java:204) ~[?:?]
      at com.mysql.cj.protocol.a.NativeProtocol.connect(NativeProtocol.java:1369) ~[?:?]
      at com.mysql.cj.NativeSession.connect(NativeSession.java:133) ~[?:?]
      at com.mysql.cj.jdbc.ConnectionImpl.connectOneTryOnly(ConnectionImpl.java:949) ~[?:?]
      at com.mysql.cj.jdbc.ConnectionImpl.createNewIO(ConnectionImpl.java:819) ~[?:?]
      at com.mysql.cj.jdbc.ConnectionImpl.<init>(ConnectionImpl.java:449) ~[?:?]
      at com.mysql.cj.jdbc.ConnectionImpl.getInstance(ConnectionImpl.java:242) ~[?:?]
      at com.mysql.cj.jdbc.NonRegisteringDriver.connect(NonRegisteringDriver.java:198) ~[?:?]
      at org.apache.flink.connector.jdbc.internal.connection.SimpleJdbcConnectionProvider.getOrEstablishConnection(SimpleJdbcConnectionProvider.java:128) ~[?:?]
      at org.apache.flink.connector.jdbc.internal.AbstractJdbcOutputFormat.open(AbstractJdbcOutputFormat.java:54) ~[?:?]
      ... 14 more

• 原因

  このエラーは通常、MySQL データベースで SSL プロトコルが有効になっているにもかかわらず、クライアントの SSL 接続が正しく設定されていない場合に発生します。たとえば、MySQL ドライバー 8.0.27 と SSL が有効な MySQL データベースの組み合わせでは、ドライバーのデフォルトのアクセス方法が SSL を使用しないため、このエラーが発生します。

• 解決策

  MySQL ディメンションテーブルの URL パラメーターに characterEncoding=utf-8&useSSL=false を追加してください。例：

  'url'='jdbc:mysql://***.***.***.***:3306/test?characterEncoding=utf-8&useSSL=false'

MySQL bigint unsigned キータイプの変更

Flink は bigint unsigned データ型をサポートしていません。データオーバーフローの可能性を回避するため、Flink は bigint unsigned のプライマリキーを decimal 型にマッピングします。Hologres への同期中は、Hologres が bigint unsigned をサポートせず、また decimal 型をプライマリキーとして許可しないため、システムはカラムを text 型に変換します。

設計および開発時に、この挙動を考慮してください。カラムを decimal 型のまま維持するには、同期を開始する前に Hologres でテーブルを手動で作成してください。このテーブルでは、別のカラムをプライマリキーとして設定するか、プライマリキーを定義しないことも可能です。ただし、この方法では元のプライマリキーでは一意性が保証されなくなるため、データの重複が発生する可能性があります。その場合は、データの重複を許容するか、重複排除ロジックを実装するなど、アプリケーションレベルでこの問題に対処する必要があります。

Flink から RDS への書き込み：更新と挿入

DDL でプライマリキーが定義されている場合、コネクタは INSERT INTO tablename(field1,field2, field3, ...) VALUES(value1, value2, value3, ...) ON DUPLICATE KEY UPDATE field1=value1,field2=value2, field3=value3, ...; ステートメントを使用します。このステートメントは、プライマリキーが存在しない場合は新しいレコードを挿入し、存在する場合は既存のレコードを更新します。DDL でプライマリキーが宣言されていない場合、コネクタは insert into ステートメントで新しいレコードを挿入します。

GROUP BY 句における一意のインデックスの使用

• ジョブの GROUP BY 句で一意のインデックスを宣言する必要があります。

• RDS テーブルが自動インクリメントプライマリキーを使用している場合、Flink ジョブではそれを PRIMARY KEY として宣言しないでください。

INT UNSIGNED のマッピング：MySQL から Flink SQL へ

MySQL JDBC ドライバーは、精度を維持するために、符号なし整数をより大きい Java データ型にマッピングします。具体的には、ドライバーは MySQL の INT UNSIGNED の値を Java LONG 型にマッピングし、Flink SQL はこれを BIGINT として扱います。同様に、ドライバーは MySQL の BIGINT UNSIGNED の値を Java BigInteger 型にマッピングし、Flink SQL はこれを DECIMAL(20, 0) として扱います。

エラー： Incorrect string value

• エラー詳細

  Caused by: java.sql.BatchUpdateException: Incorrect string value: '\xF0\x9F\x98\x80\xF0\x9F...' for column 'test' at row 1
  at sun.reflect.GeneratedConstructorAccessor59.newInstance(Unknown Source)
  at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
  at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
  at com.mysql.cj.util.Util.handleNewInstance(Util.java:192)
  at com.mysql.cj.util.Util.getInstance(Util.java:167)
  at com.mysql.cj.util.Util.getInstance(Util.java:174)
  at com.mysql.cj.jdbc.exceptions.SQLError.createBatchUpdateException(SQLError.java:224)
  at com.mysql.cj.jdbc.ClientPreparedStatement.executeBatchedInserts(ClientPreparedStatement.java:755)
  at com.mysql.cj.jdbc.ClientPreparedStatement.executeBatchInternal(ClientPreparedStatement.java:426)
  at com.mysql.cj.jdbc.StatementImpl.executeBatch(StatementImpl.java:796)
  at com.alibaba.druid.pool.DruidPooledPreparedStatement.executeBatch(DruidPooledPreparedStatement.java:565)
  at com.alibaba.ververica.connectors.rds.sink.RdsOutputFormat.executeSql(RdsOutputFormat.java:488)
  ... 15 more

• 原因

  データに特殊文字が含まれているか、データベースがサポートしていない文字エンコーディングが使用されています。

• 解決策

  JDBC 経由で MySQL データベースに接続する際、URL にcharacterEncoding=UTF-8 を追加してください。例： jdbc:mysql://<internal address>/<databaseName>?characterEncoding=UTF-8。

MySQL (TDDL/RDS) でのデッドロック

• 問題

  MySQL (TDDL/RDS) へのデータ書き込み時にデッドロックが発生します。

  重要

  Realtime Compute for Apache Flink で、MySQL などのリレーショナルデータベースをシンクとして使用する場合 (TDDL/RDS コネクタ経由)、テーブルまたはリソースへの頻繁な書き込みによってデッドロックが発生する可能性があります。

  INSERT 操作では、順番に 2 つのロック (A,B) を取得する必要があり、ロック A は範囲ロックであるとします。2 つのトランザクション (T1,T2) があり、テーブルスキーマは (id(自動インクリメントのプライマリキー),nid(一意キー)) です。T1 には 2 つのステートメント insert(null,2),(null,1) が、T2 には 1 つのステートメント insert(null,2) が含まれます。

  1. 時刻 t で、T1 は最初の INSERT ステートメントを実行します。これで T1 は両方のロック (A,B) を保持します。

  2. 時刻 t+1 に、T2 は挿入操作を開始し、範囲 (-inf,2] をロックするためにロック A を待機する必要があります。このとき、ロック A は T1 によって保持され、範囲 (-inf,2] をロックしています。範囲が包含関係にあるため、T2 は T1 による A の解放に依存します。

  3. 時刻 t+2 に、T1 は 2 番目の INSERT 文を実行し、範囲 (-inf,1] に対するロック A を要求します。この範囲は (-inf,2] の部分集合であるため、T1 はキューに入れられて、T2 がロックを解放するのを待機する必要があります。その結果、T1 は T2 によるロック A の解放に依存するようになります。

  T1 と T2 が互いにロックの解放を待ち合っているため、デッドロックが発生します。

• RDS/TDDL と Tablestore では、異なるロックメカニズムが使用されています。

  • [RDS/TDDL]： InnoDB の行ロックはインデックスに適用され、個々のレコードには適用されません。その結果、異なる行にアクセスしている場合でも、それらの行が同じインデックスキーを共有していると、ロック競合が発生する可能性があります。これにより、データ範囲全体にわたる更新が妨げられる可能性があります。

  • [Tablestore]： 単一の行ロックを使用するため、他のデータの更新には影響しません。

• デッドロックの解決策

  QPS/TPS が高い、または高同時実行の書き込みシナリオでは、デッドロックを防ぐために、結果テーブルとして Tablestore を使用してください。一般的に、Flink ジョブの結果テーブルとして TDDL または RDS を使用することは推奨されません。

  MySQL などのリレーショナルデータベースをシンクノードとして使用する必要がある場合は、次の推奨事項を検討してください。

  • 他のワークロードが同じテーブルに対して読み取りや書き込みを行っていないことを確認してください。

  • ジョブのデータ量が少ない場合は、シングルスレッド方式でデータを書き込んでみてください。ただし、QPS/TPS が高い、または高同時実行のシナリオでは、このアプローチは書き込みパフォーマンスを低下させます。

  • 一意キーを持つテーブルへの書き込みはデッドロックを引き起こす可能性があるため、可能な限り一意キーの使用は避けてください。ビジネス要件で一意キーが必要な場合は、選択性が最も高い列から最も低い列の順に列を並べて定義します。これにより、デッドロックの発生確率が大幅に低減されます。たとえば、day_time(20171010) 列の前に MD5 ハッシュを持つ列を配置することで、選択性が最も高い列が先頭になるように一意キーを定義できます。

  • ワークロードの特性に基づいてデータベースシャーディングとテーブル分割を使用し、複数のテーブルに書き込みを分散させてください。実装の詳細については、データベース管理者にお問い合わせください。

ダウンストリームテーブル構造の更新失敗

テーブル構造の同期では、DDL ステートメントは追跡されません。代わりに、連続するレコードを比較することでスキーマ変更が検出されます。DDL 変更が発生した後にアップストリームでデータ変更がない場合、ダウンストリームのテーブル構造は更新されません。詳細については、「テーブル構造の変更の同期ポリシー」をご参照ください。

スプリット完了応答のタイムアウトエラー

このエラーは、タスクの CPU 使用率が高いために、コーディネーターからの RPC リクエストに応答できない場合に発生します。この問題を解決するには、リソース設定ページで TaskManager の CPU リソースを増やしてください。

フルロード中のスキーマ変更の影響

フルロードフェーズ中にスキーマが変更されると、ジョブが失敗したり、スキーマ変更が同期されなくなったりする可能性があります。この問題を解決するには、ジョブを停止し、ダウンストリームテーブルを削除してから、ジョブを状態なしで再起動してください。

CTAS/CDAS 同期におけるサポート対象外のスキーマ変更

テーブルのデータを再同期してください。そのためには、ジョブを停止し、ダウンストリームテーブルを削除してから、ステートレススタートで同期ジョブを再起動します。このような互換性のない変更を行うと、再起動時にジョブが再び失敗するため、避けてください。サポートされているスキーマ変更の詳細については、「CREATE TABLE AS (CTAS) ステートメント」をご参照ください。

ClickHouse でのリトラクション更新

Flink 結果テーブルの DDL でプライマリキーを指定し、ignoreDelete パラメーターを false に設定すると、ClickHouse 結果テーブルでリトラクション更新が利用可能になります。ただし、これによりパフォーマンスが大幅に低下します。

ClickHouse は、オンライン分析処理 (OLAP) 用に設計されたカラムナデータベース管理システムで、UPDATE および DELETE 操作のサポートは限定的です。Flink DDL でプライマリキーを指定すると、コネクタはALTER TABLE UPDATE およびALTER TABLE DELETE を使用してデータを更新および削除しようとします。これらの操作は非常に非効率的です。

ClickHouse でのデータの可視性

• exactly-once セマンティクス が無効 (デフォルト) な ClickHouse の結果テーブルでは、データはバッファーがフラッシュされると可視になります。システムは、レコード数が batchSize の値に達するか、最後の書き込みからの時間が flushIntervalMs を超えると、このバッファーを自動的にフラッシュします。チェックポイント の完了を待つ必要はありません。

• exactly-once セマンティクス が有効な ClickHouse の結果テーブルでは、データは対応する チェックポイント が正常に完了した後にのみ可視になります。

出力結果の表示

ディメンションテーブルの結合でデータが返されない

DDL ステートメント内のスキーマ (データ型や列名など) が、物理テーブルのスキーマと一致していることを確認してください。

max_pt() と max_pt_with_done()

Paimon 書き込みジョブエラー：「Heartbeat of TaskManager timed out」

このエラーの最も可能性の高い原因は、TaskManager のヒープメモリ不足です。Paimon は主に次の方法でヒープメモリを使用します。

• Paimon プライマリキーテーブルの書き込みオペレーターの各並列インスタンスには、ソート用のメモリバッファーがあります。このバッファーのサイズは write-buffer-size テーブルプロパティによって制御され、デフォルト値は 256 MB です。

• Paimon はデフォルトで ORC ファイルフォーマットを使用します。このフォーマットでは、メモリ内のデータをバッチでカラムナフォーマットに変換するための追加のメモリバッファーが必要です。このバッファーのサイズは orc.write.batch-size テーブルプロパティによって制御され、デフォルト値は 1024 です。つまり、バッファーは 1024 行のデータを保持します。

• 変更された各バケットには、そのデータを書き込むための専用の書き込みオブジェクトがあります。

これらの使用パターンに基づき、ヒープメモリ不足の潜在的な原因とその解決策を以下に示します。

• write-buffer-size の値が大きすぎる。

  このパラメーターを減らしてみてください。ただし、バッファーが小さすぎると、ディスク書き込みが頻繁に発生し、スモールファイルのコンパクションがより頻繁にトリガーされ、書き込みパフォーマンスに影響を与える可能性があります。

• 単一のデータレコードが大きすぎる。

  たとえば、レコードに 4 MB の JSON フィールドが含まれている場合、ORC バッファーは 4 MB × 1024 = 4 GB まで増加し、大量のヒープメモリを消費する可能性があります。次の 2 つの解決策があります。

  • orc.write.batch-size の値を減らします。

  • Paimon の結果テーブルに対してアドホッククエリ (OLAP) を実行する必要がなく、バッチ消費またはストリーミング消費のみが必要な場合は、テーブル作成時に 'file.format' = 'avro' および 'metadata.stats-mode' = 'none' テーブルプロパティを設定できます。これにより、テーブルが Avro フォーマットに切り替わり、統計情報収集が無効になります。

    説明

    これらのパラメーターは、テーブル作成時にのみ設定できます。テーブル作成後に ALTER TABLE 文や SQL ヒントを使用して変更することはできません。

• 同時に多数のパーティションに書き込んだり、パーティションあたりのバケット数が多すぎたりすると、書き込みオブジェクトの数が過剰になります。

  パーティション列の設定を見直して、適切かどうかを確認してください。誤った SQL によって予期しないデータがパーティション列に書き込まれていないことを確認してください。また、バケット数が妥当かどうかも確認してください。ベストプラクティスとして、バケットあたりの合計データサイズは約 2 GB とし、5 GB を超えないようにする必要があります。バケット数の調整の詳細については、「固定バケットテーブルのバケット数の調整」をご参照ください。

エラー： "Sink materializer must not be used with Paimon sink"

シンクマテリアライザーオペレーターは、ストリーミングジョブのカスケード結合から生じる順序不同のデータを処理します。しかし、Paimon テーブルに書き込むジョブでは、このオペレーターはオーバーヘッドを発生させ、集約を使用する場合に不正な結果を引き起こす可能性があります。Paimon シンクでシンクマテリアライザーオペレーターを使用しないでください。

シンクマテリアライザーオペレーターは、SET 文またはランタイムパラメーターを使用して table.exec.sink.upsert-materialize パラメーターを false に設定することで無効にできます。順序不同のデータを処理する必要もある場合は、「順序不同のデータの処理」をご参照ください。

Paimon： File deletion conflicts detected または LSM conflicts detected エラー

このエラーは、次のいずれかの理由で発生する可能性があります。

• 複数のジョブが同じ Paimon テーブルの同じパーティションに書き込んでいる。この場合、Paimon は フェールオーバーと再起動 によって競合を解決します。これは想定されている動作であり、エラーが再発しない限りアクションは不要です。

• ジョブが古い状態から復元されたため、エラーが再発します。この問題を解決するには、ジョブを最新の状態から復元するか、 ステートなしで開始 してください。

• Paimon は、単一のジョブ内の複数の INSERT 文からの個別の書き込みをサポートしていません。代わりに、 UNION ALL 文を使用して、複数のデータストリームを Paimon テーブルに書き込んでください。

• Global Committer ノードまたは Compaction Coordinator ノード (Append Scalable テーブル への書き込み時) の同時実行数が 1 より大きいです。これらのノードの 同時実行数 は、 データ整合性 を確保するために 1 である必要があります。

Paimon 消費ジョブにおける "File xxx not found" エラー

Paimon テーブルの消費は スナップショットファイル に依存します。スナップショットの保持期間が短すぎる場合、または消費ジョブの効率が低い場合、ジョブが完了する前にスナップショットファイルの有効期限が切れて削除される可能性があります。これにより、消費ジョブが失敗します。

この問題を解決するには、スナップショットファイルの保持期間を調整するか、コンシューマー ID を指定するか、または消費ジョブを最適化します。使用可能なスナップショットファイルとその作成タイムスタンプを確認するには、Snapshots システムテーブルをご参照ください。

Paimon ジョブのエラー： No space left on device

• ジョブが Paimon クエリを実行する場合 (Paimon テーブルをディメンションテーブルとして使用する、または changelog-producer='lookup' を設定するなど)、過剰な数のキャッシュファイルがこのエラーを引き起こす可能性があります。これを防ぐには、SQL ヒントを使用して次のパラメーターを設定し、クエリキャッシュの最大ディスク領域と保持期間を制限します。

  • lookup.cache-max-disk-size： クエリキャッシュが使用できる最大ローカルディスク領域。推奨値には、256 MB、512 MB、1 GB などがあります。

  • lookup.cache-file-retention： クエリキャッシュファイルの保持期間。推奨値には、30 分、15 分、またはそれより短い時間があります。

• Paimon テーブルに書き込むジョブの場合は、SQL ヒントを使用して次のパラメーターを設定します。これらの設定により、書き込みプロセス中のローカル一時ファイルのサイズを制限し、ディスク領域不足を防ぎます。

  • write-buffer-spillable： 書き込みバッファーがディスクにスピルできるかどうかを制御します。これを false に設定すると、バッファーがディスク領域を一切使用しなくなります。

  • write-buffer-spill.max-disk-size： 書き込みバッファーがスピル時に使用できる最大ディスク領域。推奨値には、256 MB、512 MB、1 GB などがあります。

OSS 上の Paimon ファイルの管理

• Paimon は、テーブルの以前のバージョンにアクセスするために履歴データファイルを保持します。これらのファイルの保持ポリシーを調整することで、ストレージを管理できます。詳細な手順については、「期限切れデータのクリーンアップ」をご参照ください。

• 不適切なパーティション列の設定や、過剰な数のバケットも、この問題を引き起こす可能性があります。ベストプラクティスとして、バケットあたりのデータサイズは、約 2 GB、最大で 5 GB を目標としてください。詳細については、「バケット化」をご参照ください。

• デフォルトでは、データファイルは ORC フォーマットで保存されます。データファイルの合計サイズを削減するには、テーブル作成時にテーブルパラメーター 'file.compression' = 'zstd' を設定することで、ZSTD 圧縮フォーマットを使用できます。

  説明

  このパラメーターはテーブル作成時にのみ設定でき、後から ALTER TABLE 文または SQL ヒントで変更することはできません。

データの可視性はチェックポイント間隔に依存

はい。 Paimon は exactly-once セマンティクスを保証するためにチェックポイントに依存しています。データは、チェックポイントが完了したときにのみコミットされ、ダウンストリームで参照可能になります。このコミットが行われる前は、ローカルバッファー内のデータはリモートファイルシステムにフラッシュされますが、まだ読み取り可能ではありません。

長時間実行される Paimon ジョブでのメモリ使用量の緩やかな増加

• ジョブの rps も緩やかに増加している場合、メモリ使用量の増加は想定される動作です。

• Paimon のファイルシステムカタログを使用して OSS の読み取りまたは書き込みを行う場合は、fs.oss.endpoint、fs.oss.accessKeyId、fs.oss.accessKeySecret のカタログパラメーターを設定してください。設定しない場合、Flink ジョブで緩やかなメモリリークが発生する可能性があります。これはコミュニティでは既知のイシューです。

IllegalArgumentException: timeout value is negative

• エラー詳細

  2021-02-24 15:14:58
  java.lang.RuntimeException: java.lang.RuntimeException: java.lang.IllegalArgumentException: timeout value is negative
      at com.alibaba.ververica.connectors.common.source.reader.ParallelReader.run(ParallelReader.java:166)
      at com.alibaba.ververica.connectors.common.source.AbstractParallelSourceBase.run(AbstractParallelSourceBase.java:205)
      at com.alibaba.ververica.connectors.metaq.source.MetaQRowDataSource.run(MetaQRowDataSource.java:84)
      at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:100)
      at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:63)
      at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:213)
  Caused by: java.lang.RuntimeException: java.lang.IllegalArgumentException: timeout value is negative
      at com.alibaba.ververica.connectors.common.source.reader.ParallelReader.runImpl(ParallelReader.java:244)

• エラーの原因

  新しい MQ メッセージが消費されない場合、MetaQSource スレッドは pullIntervalMs パラメーターで定義された間隔でスリープします。このパラメーターのデフォルト値は -1 です。スリープ時間は負の値にできないため、ジョブは IllegalArgumentException で失敗します。

• 解決策

  pullIntervalMs パラメーターを 0 以上の値に設定してください。

パーティション変更検出

• Realtime Compute for Apache Flink 6.0.2 より前のバージョンでは、ソースオペレーターが 5～10 分ごとにパーティション数を取得します。パーティション数が 3 回連続のチェックで異なる場合、フェールオーバーが発生します。その結果、ソースは 10～30 分以内にフェールオーバーを開始します。ジョブの再起動後、更新されたパーティションセットから読み取ります。

• Realtime Compute for Apache Flink 6.0.2 以降のバージョンでは、ソースオペレーターがデフォルトで 5 分ごとにパーティション数を取得します。新しいパーティションが検出されると、TaskManager 上のソースオペレーターに直接割り当てられ、ソースオペレーターがデータの読み取りを開始します。ジョブのフェールオーバーは不要なため、ソースは 1～5 分以内にパーティション変更を検出できます。

エラー： バックプレッシャーがリジェクトリミットを超過

• エラー詳細

  26      at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:47)
  27      at org.apache.flink.streaming.runtime.tasks.StreamTask.performCheckpoint(StreamTask.java:911)
  28      at org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpointOnBarrier(StreamTask.java:879)
  29      ... 13 more
  30  Caused by: java.lang.RuntimeException: Rpc Exception failed errorCount=6 with RpcException: request niagara.table.proto.UpsertRecordBatchRequest@b53e2d74 failed on final try 4, maxAttempts=4, sn=11.117.xxx, errorCode=11, msg=BackPresure Exceed Reject Limit [method:UpsertRecordBatch,transaction_id:xxx,table_id:xxx,table_version:128, actor_id:74538 xxx,worker_address:11.117.xxx]
  31          at com.alibaba.ververica.connectors.hologres.sink.HologresOutputFormat.sync(HologresOutputFormat.java:264)
  32          at com.alibaba.ververica.connectors.common.sink.OutputFormatSinkFunction.snapshotState(OutputFormatSinkFunction.java:91)
  33          at org.apache.flink.streaming.util.functions.StreamingFunctionUtils.trySnapshotFunctionState(StreamingFunctionUtils.java:128)
  34          at org.apache.flink.streaming.util.functions.StreamingFunctionUtils.snapshotFunctionState(StreamingFunctionUtils.java:101)
  35          at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.snapshotState(AbstractUdfStreamOperator.java:90)
  36          at org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.snapshotState(StreamOperatorStateHandler.java:186)
  37          ... 23 more

• 原因

  Hologres インスタンスの書き込み負荷が高すぎます。

• 解決策

  インスタンス情報を添えて Hologres のテクニカルサポートに連絡し、アップグレードを依頼してください。

エラー： remaining connection slots are reserved for non-replication superuser connections

• エラー詳細

  Caused by: com.alibaba.hologres.client.exception.HoloClientWithDetailsException: failed records 1, first:Record{schema=org.postgresql.model.TableSchema@188365, values=[f06b41455c694d24a18d0552b8b0****, com.chot.tpfymnq.meta, 2022-04-02 19:46:40.0, 28, 1, null], bitSet={0, 1, 2, 3, 4}},first err:[106]FATAL: remaining connection slots are reserved for non-replication superuser connections
      at com.alibaba.hologres.client.impl.Worker.handlePutAction(Worker.java:406) ~[?:?]
      at com.alibaba.hologres.client.impl.Worker.run(Worker.java:118) ~[?:?]
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[?:1.8.0_302]
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[?:1.8.0_302]
      ... 1 more
  Caused by: com.alibaba.hologres.org.postgresql.util.PSQLException: FATAL: remaining connection slots are reserved for non-replication superuser connections
      at com.alibaba.hologres.org.postgresql.core.v3.QueryExecutorImpl.receiveErrorResponse(QueryExecutorImpl.java:2553) ~[?:?]
      at com.alibaba.hologres.org.postgresql.core.v3.QueryExecutorImpl.readStartupMessages(QueryExecutorImpl.java:2665) ~[?:?]
      at com.alibaba.hologres.org.postgresql.core.v3.QueryExecutorImpl.<init>(QueryExecutorImpl.java:147) ~[?:?]
      at com.alibaba.hologres.org.postgresql.core.v3.ConnectionFactoryImpl.openConnectionImpl(ConnectionFactoryImpl.java:273) ~[?:?]
      at com.alibaba.hologres.org.postgresql.core.ConnectionFactory.openConnection(ConnectionFactory.java:51) ~[?:?]
      at com.alibaba.hologres.org.postgresql.jdbc.PgConnection.<init>(PgConnection.java:240) ~[?:?]
      at com.alibaba.hologres.org.postgresql.Driver.makeConnection(Driver.java:478) ~[?:?]
      at com.alibaba.hologres.org.postgresql.Driver.connect(Driver.java:277) ~[?:?]
      at java.sql.DriverManager.getConnection(DriverManager.java:674) ~[?:1.8.0_302]
      at java.sql.DriverManager.getConnection(DriverManager.java:217) ~[?:1.8.0_302]
      at com.alibaba.hologres.client.impl.ConnectionHolder.buildConnection(ConnectionHolder.java:122) ~[?:?]
      at com.alibaba.hologres.client.impl.ConnectionHolder.retryExecute(ConnectionHolder.java:195) ~[?:?]
      at com.alibaba.hologres.client.impl.ConnectionHolder.retryExecute(ConnectionHolder.java:184) ~[?:?]
      at com.alibaba.hologres.client.impl.Worker.doHandlePutAction(Worker.java:460) ~[?:?]
      at com.alibaba.hologres.client.impl.Worker.handlePutAction(Worker.java:389) ~[?:?]
      at com.alibaba.hologres.client.impl.Worker.run(Worker.java:118) ~[?:?]
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[?:1.8.0_302]
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[?:1.8.0_302]
      ... 1 more

• 原因

  Hologres インスタンスの接続上限を超えました。

• 解決策

  • 各フロントエンド (FE) への接続の app_name を確認し、Flink コネクタからの Hologres クライアント接続数をカウントします。

  • Hologres に接続している他のジョブを確認します。

  • 接続を解放します。詳細については、「接続管理」をご参照ください。

no table is defined in publication

• エラー詳細

  同じ名前のテーブルを削除して再作成すると、ジョブで no table is defined in publication というエラーが報告されることがあります。

• 原因

  テーブルを削除しても、関連付けられたパブリケーションは削除されません。

• 解決策

  1. Hologres で select * from pg_publication where pubname not in (select pubname from pg_publication_tables); コマンドを実行して、テーブルが削除された際にクリーンアップされなかったパブリケーション情報を照会します。

  2. drop publication xx; ステートメントを実行して、残っているパブリケーションを削除します。

  3. ジョブを再起動します。

チェックポイント間隔とデータの可視性

Flink Hologres sink コネクタのチェックポイント間隔は、Hologres におけるデータの可視性を直接制御しません。主な役割は、障害復旧の SLA を定義することです。

Hologres コネクタはトランザクションをサポートしていません。インメモリ バッファーを定期的にデータベースへフラッシュします。チェックポイントは、その完了までにすべてのデータがフラッシュされることを保証しますが、コネクタは間隔全体が経過するまでフラッシュを待機しません。特定のバッファー条件が満たされると、コネクタは早めにフラッシュをトリガーします (詳細については、Hologres、Hologres、Hologres をご参照ください)。データウェアハウスでは通常、トランザクション整合性が不要であるため、コネクタはバックグラウンドで非同期にデータをフラッシュします。その後、障害復旧に備えるため、各チェックポイント中に最終的な強制フラッシュを実行します。

ジョブのデプロイメントにおける permission denied for database エラー

• 原因

  Realtime Compute for Apache Flink VVR 8.0.4 以降、コネクタは Hologres インスタンス V2.0 以降からバイナリログを消費するために JDBC モードを強制します。これらのインスタンスでは、非スーパーユーザー アカウントが JDBC モードでバイナリログを消費するには特別な権限が必要です。

• 解決策

  非スーパーユーザー アカウントに、JDBC モードでバイナリログを消費する権限を付与してください。

  [user_name] は、Alibaba Cloud アカウント ID または RAM ユーザーを指します。詳細については、「アカウント概要」をご参照ください。

  -- エキスパート権限モデルの場合、ユーザーに CREATE 権限とレプリケーションロールを付与してください。
  GRANT CREATE ON DATABASE <db_name> TO <user_name>;
  alter role <user_name> replication;
  -- データベースがシンプル権限モデル (SPM) を使用している場合、GRANT ステートメントは実行できません。
  -- 代わりに、spm_grant を使用してユーザーにデータベースの Admin ロールを付与してください。HoloWeb で直接権限を付与することもできます。
  call spm_grant('<db_name>_admin', '<user_name>');
  alter role <user_name> replication;

ジョブのリカバリ失敗： table id parsed from checkpoint is different from the current table id

• 原因

  この例外は、Realtime Compute for Apache Flink バージョン VVR 8.0.5 から VVR 8.0.8 で発生します。Hologres バイナリログソーステーブルを使用しているジョブがチェックポイントからリカバリする際、エンジンはテーブル ID に対して厳密なチェックを実施します。Hologres テーブルの現在のテーブル ID がチェックポイントに保存されているものと一致しない場合、リカバリは失敗します。これは、ジョブの実行中にソーステーブルが切り捨てられたか、再作成されたことを示しています。

• 解決策

  VVR 8.0.9 以降のバージョンにアップグレードし、ジョブを再起動してください。VVR 8.0.9 では、複雑なビジネスシナリオに対応するため、テーブル ID の厳密なチェックが削除されています。ただし、バイナリログソーステーブルの再作成は避けてください。テーブルが再作成されると、そのバイナリログ履歴全体がクリアされます。Flink が古いテーブルのコンシューマーオフセットを使用して新しいテーブルからデータを読み取ると、データの不整合が発生する可能性があります。

JDBC モードにおける想定外の Binlog データ精度

• 原因

  Realtime Compute for Apache Flink 8.0.10 以前では、Binlog ソーステーブルの Flink DDL で宣言したDECIMAL 型の精度が Hologres の精度と一致しない場合、データ精度が想定外のものとなります。

• ソリューション

  この問題は Realtime Compute for Apache Flink 8.0.11 で修正されています。ただし、精度損失を防ぐために、Flink と Hologres の間でDECIMAL 型の精度が一致していることを確認してください。

同じ名前のテーブルを削除して再作成すると、ジョブで no table is defined in publication または The table xxx has no slot named xxx の例外が報告される問題

• 原因

  これは、テーブルを削除しても、関連するパブリケーションが削除されないために発生します。

• 解決策

  解決策 1： Hologres で select * from pg_publication where pubname not in (select pubname from pg_publication_tables); 文を実行して、削除されたテーブルから残されたパブリケーションを見つけます。次に、drop publication xx; 文を実行してそれらを削除します。最後に、ジョブを再起動します。

  解決策 2： VVR 8.0.5 以降を使用してください。コネクタが自動的にクリーンアップを処理します。

Hologres からの読み取り時の ClassCastException

• エラー詳細

  エラーメッセージは次のようなものです。

  java.lang.ClassCastException: class java.lang.Integer cannot be cast to class java.lang.Long (java.lang.Integer and java.lang.Long are in module java.base of loader 'bootstrap')

• 原因

  このエラーは、Flink DDL のフィールドの型が Hologres 物理テーブルの対応するフィールドの型と一致しない場合に発生します。たとえば、Flink DDL でフィールドが BIGINT として定義されているのに、Hologres テーブルの対応するフィールドが INTEGER である場合です。NULL 値に対しては型チェックがスキップされるため、ジョブは実際のデータを読み取ったときにのみこの例外をスローします。

• 解決策

  Hologres の データ型まとめ ドキュメントを参照し、Flink DDL のフィールド型が Hologres 物理テーブルのフィールド型と一致していることを確認してください。

LogSizeTooLargeException

• エラーの詳細

  Caused by: com.aliyun.openservices.aliyun.log.producer.errors.LogSizeTooLargeException: the logs is 8785684 bytes which is larger than MAX_BATCH_SIZE_IN_BYTES 8388608
  at com.aliyun.openservices.aliyun.log.producer.internals.LogAccumulator.ensureValidLogSize(LogAccumulator.java:249)
  at com.aliyun.openservices.aliyun.log.producer.internals.LogAccumulator.doAppend(LogAccumulator.java:103)
  at com.aliyun.openservices.aliyun.log.producer.internals.LogAccumulator.append(LogAccumulator.java:84)
  at com.aliyun.openservices.aliyun.log.producer.LogProducer.send(LogProducer.java:385)
  at com.aliyun.openservices.aliyun.log.producer.LogProducer.send(LogProducer.java:308)
  at com.aliyun.openservices.aliyun.log.producer.LogProducer.send(LogProducer.java:211)
  at com.alibaba.ververica.connectors.sls.sink.SLSOutputFormat.writeRecord(SLSOutputFo
  rmat.java:100)

• 原因

  このエラーは、Log Service に送信されたログのバッチが 8 MB のサイズ制限を超えているためです。

• 解決策

  サイズ超過のログエントリをスキップするには、起動位置を変更します。詳細については、「ジョブの起動」をご参照ください。

リカバリ中の TaskManager OOM： Java ヒープスペースエラー

• 原因

  この問題は通常、SLS メッセージ本文が大きすぎることが原因で発生します。SLS コネクタはデータをバッチでリクエストします。バッチあたりの LogGroup 数は batchGetSize パラメーターで制御され、デフォルト値は 100 です。つまり、1 回のリクエストで最大 100 個の LogGroup を取得できます。通常時、Flink プログラムはデータを速やかに消費するため、100 個の LogGroup をフルバッチで取得することはほとんどありません。しかし、フェールオーバー中は、未消費のデータが大量に蓄積される可能性があります。100 個の LogGroup からなるフルバッチに必要なメモリが JVM の使用可能なメモリを超える場合、TaskManager で OOM が発生します。

• 解決策

  batchGetSize パラメーターの値を小さくしてください。

Paimon ソーステーブルの消費オフセット設定

Paimon ソーステーブルの消費オフセットを設定するには、scan.mode パラメーターを使用します。次の表に、使用可能な値とその動作を示します。

自動パーティションの有効期限の設定

Paimon テーブルでは、存続時間が指定されたパーティションの有効期限を超えたパーティションを自動的に削除できます。この機能はストレージコストの削減に役立ちます。プロセスは次のとおりです：

• 存続時間：現在のシステム時刻から、パーティション値から導出したタイムスタンプを差し引いた値です。パーティション値は次の手順でタイムスタンプに変換されます：

  1. partition.timestamp-pattern パラメーターで指定されたフォーマット文字列を使用して、パーティション値を時刻文字列に変換します。

     このフォーマット文字列では、パーティション列はドル記号 ($) に続けて列名で表されます。例えば、パーティション列が year、month、day、hour の場合、フォーマット文字列 $year-$month-$day $hour:00:00 によって、パーティション year=2023,month=04,day=21,hour=17 は文字列 2023-04-21 17:00:00 に変換されます。

  2. partition.timestamp-formatter パラメーターで指定されたフォーマット文字列を使用して、時刻文字列をタイムスタンプに変換します。

     このパラメーターが設定されていない場合、システムはデフォルトで yyyy-MM-dd HH:mm:ss および yyyy-MM-dd フォーマットを使用します。Java の DateTimeFormatter と互換性のある任意のフォーマット文字列を使用できます。

• パーティションの有効期限：partition.expiration-time パラメーターに設定する値です。

ストレージにデータが見つからない場合のトラブルシューティング

• データはストレージにすぐには表示されない場合があります。Flink ライターは、次の条件下でデータをディスクにフラッシュします。

  • バケット内でバッファーされたデータが一定のサイズに達した場合 （デフォルト： 64 MB）。

  • バッファーサイズの合計がしきい値に達した場合 （デフォルト： 1 GB）。

  • チェックポイントがトリガーされてすべてのインメモリ データがフラッシュされる場合。

• ストリーム書き込みを使用している場合は、チェックポイントが有効になっていることを確認してください。

Hudi での重複データの処理

• COW 書き込みの場合、write.insert.drop.duplicates パラメーターを有効にします。

  デフォルトでは、COW 書き込みは各バケットの最初のファイル内のデータを重複排除せず、増分データにのみ重複排除を適用します。グローバルな重複排除を実行するには、このパラメーターを有効にする必要があります。MOR 書き込みの場合、追加のパラメーターは必要ありません。プライマリキーを定義すると、デフォルトでグローバルな重複排除が有効になります。

  説明

  Hudi バージョン 0.10.0 以降、このプロパティは write.precombine に名前が変更され、デフォルトで true に設定されています。

• 複数のパーティションにわたって重複排除を実行するには、index.global.enabled パラメーターを true に設定します。

  説明

  • Hudi バージョン 0.10.0 以降、このプロパティはデフォルトで true に設定されています。

  • index.type=bucket の場合、index.global.enabled パラメーターを true に設定しても効果がありません。バケットインデックスはクロスパーティション変更をサポートしていないためです。したがって、グローバルインデックスが有効になっている場合でも、重複排除機能は複数のパーティションにわたって機能しません。

• 1 か月前のデータの変更など、ロングウィンドウ更新の場合は、日数で表される index.state.ttl パラメーターを増やします。

  インデックスは、重複データを識別するための Hudi のコアデータ構造です。index.state.ttl パラメーターは、インデックスの状態が保持される期間を制御します。以前のデフォルト値は 1.5 日でした。0 以下の値は、インデックスの状態が永続的に保持されることを示します。

  説明

  Hudi バージョン 0.10.0 以降、このプロパティはデフォルトで 0 に設定されています。

Merge On Read でのログファイルのみの生成

• 原因：Hudi はコンパクション後にのみ Parquet ファイルを作成し、それ以外の場合はログファイルのみを作成します。デフォルトでは、Merge On Read テーブルは非同期コンパクションを使用し、5 回のコミットごとにコンパクションジョブがトリガーされます。

• 解決策：より早いタイミングでコンパクションジョブをトリガーするには、compaction.delta_commits パラメーターを調整してください。

エラー：「multi-statement be found」

• 問題

  AnalyticDB for MySQL (ADB) インスタンスにデータを書き込む Flink ジョブが失敗して再起動します。ログには、次のようなエラーが表示されます： Caused by: java.sql.SQLSyntaxErrorException: [13000, 2024101216171419216823505703151806929] multi-statement be found.

  at java.util.TimerThread.run(Timer.java:505)
  Caused by: java.sql.BatchUpdateException: [13000, 2024101216400819216823505703151079281] multi-statement be found.
  	at sun.reflect.GeneratedConstructorAccessor115.newInstance(Unknown Source)
  	at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
  	at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
  	at com.mysql.cj.util.Util.handleNewInstance(Util.java:192)
  	at com.mysql.cj.util.Util.getInstance(Util.java:167)
  	at com.mysql.cj.util.Util.getInstance(Util.java:174)
  	at com.mysql.cj.jdbc.exceptions.SQLError.createBatchUpdateException(SQLError.java:224)
  	at com.mysql.cj.jdbc.ClientPreparedStatement.executePreparedBatchAsMultiStatement(ClientPreparedStatement.java:584)
  	at com.mysql.cj.jdbc.ClientPreparedStatement.executeBatchInternal(ClientPreparedStatement.java:431)
  	at com.mysql.cj.jdbc.StatementImpl.executeBatch(StatementImpl.java:795)
  	at com.ververica.cdc.connectors.shaded.com.zaxxer.hikari.pool.ProxyStatement.executeBatch(ProxyStatement.java:127)
  	at com.ververica.cdc.connectors.shaded.com.zaxxer.hikari.pool.HikariProxyPreparedStatement.executeBatch(HikariProxyPreparedStatement...)
  	at com.ververica.connectors.mysql.table.sink.MySqlOutputFormat.executeSql(MySqlOutputFormat.java:567)
  	... 6 more

• 原因

  このエラーは、バージョン 8.x の MySQL JDBC ドライバーと、allowMultiQueries=true パラメーターが有効になっている AnalyticDB for MySQL (ADB) データベースとの間の互換性の問題に起因します。

• ソリューション

  1. テクニカルサポートに問い合わせて、バージョン 5.1.46 の MySQL JDBC ドライバーを使用するカスタム ADB 3.0 コネクタを入手してください。このコネクタを Flink タスクに適用してください。カスタムコネクタの使用方法については、「カスタムコネクタの管理」をご参照ください。

  2. ADB テーブルの URI で allowMultiQueries=true パラメーターを設定してください。例：jdbc:mysql://xxxxx.ads.aliyuncs.com:3306/xxx?allowMultiQueries=true。

エラー： No suitable driver found

• 原因

  カスタムコネクタが必要なドライバーを見つけられません。

• 解決策

  • Class.forName を呼び出して、ファクトリークラスでドライバーをロードします。

  • ドライバーを追加の依存関係として追加し、kubernetes.application-mode.classpath.include-user-jar パラメーターを true に設定します。手順については、「カスタムジョブ実行パラメーターの設定方法」をご参照ください。

Flink から Elasticsearch に書き込む際のデータ損失または上書き

• 原因 1： doc_as_upsert と Elasticsearch インジェストパイプラインの競合

  次の両方を同時に設定すると、Elasticsearch は部分更新とパイプライン変換を互換性のない順序で処理し、ドキュメントが想定外に上書きされたり失われたりする可能性があります：

  • Flink DDL の WITH 句における sink.bulk-flush.update.doc_as_upsert = 'true' の設定

  • ターゲットインデックスに割り当てられた Elasticsearch インジェストパイプライン

  この構成では、Elasticsearch は部分更新を処理する前にインジェストパイプラインを適用します。パイプラインロジックと Elasticsearch バージョンによっては、この相互作用により、ドキュメントフィールドが誤った値で上書きされたり、ドキュメントが通知なしにドロップされたりする可能性があります。

• 原因 2： 複数の Flink シンクタスクによる同じインデックスへの書き込み

  2 つ以上の Flink ジョブ、または複数の並列シンクインスタンスが、ドキュメントキーの割り当てを調整せずに同じ Elasticsearch インデックスに書き込むと、競合状態が発生し、ドキュメントが互いに上書きされる可能性があります。その結果、特定のドキュメント ID に対して最後に書き込んだシンクのデータ以外は失われます。

• 解決策

  • doc_as_upsert とインジェストパイプラインの競合に対する解決策： Flink DDL から sink.bulk-flush.update.doc_as_upsert = 'true' パラメーターを削除し、ターゲットインデックスから Elasticsearch インジェストパイプラインを削除するか、再割り当てします。パイプラインで処理していたデータ変換ロジックは、Flink ジョブ自体に移します。例えば、シンクオペレーターの前で ProcessFunction や計算列を使用します。

  • 複数のシンクタスクによる相互上書きに対する解決策： 各 Flink ジョブが別々の Elasticsearch インデックスに書き込むようにするか、ドキュメントキーを決定的に管理する単一の Flink ジョブに書き込みを統合してください。