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

典型的なシナリオ

制限事項と注意事項

この機能を使用する前に、その制限事項と潜在的な脅威を理解する必要があります：

• コネクタのサポート：現在、この機能をサポートしているのは Kafka データソースのみです。他のソースへのサポートは順次追加されています。

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

構文

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

ダーティデータコレクターは、パイプラインモジュールで定義されます。構文は次のとおりです：

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

データソースでのフォールトトレランスポリシーの設定

ダーティデータ収集を設定するだけでは、解析エラーのスキップは有効になりません。この機能は、Kafka のフォールトトレランスポリシーと組み合わせて使用する必要があります。詳細については、「Kafka コネクタドキュメント」をご参照ください。次の例は、設定のサンプルを示しています：

source:
  type: kafka
  # Skip the first 100 parsing exceptions. If the number of exceptions exceeds 100, the job fails.
  ingestion.ignore-errors: true
  ingestion.error-tolerance.max-count: 100

Logger ダーティデータコレクター

Logger ダーティデータコレクターは、ダーティデータを別のログファイルに保存します。ダーティデータのログファイルを表示するには、次の手順を実行します：

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

2. [運用ログ] をクリックし、[実行中の TaskManager] サブタブに移動して、オペレーター用のテクノロジーマネージャー (TM) ノードを選択します。

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