Realtime Compute for Apache Flink:ダーティデータの収集

代表的なユースケース

制限事項と考慮事項

この機能を使用する前に、制限事項と潜在的なリスクを理解してください。

• サポート対象のコネクター：この機能は現在、Kafka データソースでのみ利用可能です。他のソースへの対応は、今後のリリースで追加される予定です。

• サポート対象のコレクタータイプ：現在、logger タイプのみがサポートされており、ダーティデータをログファイルに書き込みます。

構文

ダーティデータコレクターの有効化

ダーティデータコレクターは pipeline モジュールで定義します。構文は以下のとおりです。

pipeline:
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

データソースでのエラー耐性の設定

ダーティデータコレクターを設定するだけでは、解析エラーをスキップできません。この機能は、Kafka エラー耐性ポリシーと組み合わせて使用する必要があります。詳細については、Kafka コネクタードキュメントをご参照ください。以下は一般的な設定例です。

source:
  type: kafka
  # 最初の 100 件の解析エラーをスキップします。100 件を超えるとジョブは失敗します。
  ingestion.ignore-errors: true
  ingestion.error-tolerance.max-count: 100

ロガーダーティデータコレクター

ロガーダーティデータコレクターは、ダーティデータを個別のログファイルに保存します。ダーティデータログを表示するには、次の手順に従ってください。

1. [運用保守] ページに移動し、[ジョブログ] タブをクリックします。

2. [実行ログ] をクリックし、[実行中の TaskManager] サブタブを選択してから、該当するオペレーターの TaskManager ノードを選択します。

3. [ログリスト] をクリックし、リスト内の yaml-dirty-data.out という名前のログファイルをクリックして、ダーティデータレコードを表示または保存します。

各ダーティデータレコードには、以下のメタデータが含まれます。

• ダーティデータが処理されたタイムスタンプ

• ダーティデータレコードを出力したオペレーターとサブタスクインデックス

• 生データの内容

• 処理失敗の原因となった例外情報

text[2025-04-05 10:23:45] [Operator: SourceKafka -> Subtask: 2]
Raw Data: {"id": "abc", "ts": "invalid-timestamp"}
Exception: java.time.format.DateTimeParseException: Text 'invalid-timestamp' could not be parsed at index 0
---