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

典型的な利用シーン

制限事項と注意事項

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

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

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

構文構造

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

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

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

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

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

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

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

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

1. [ジョブの O&M] ページに移動し、[ジョブログ] タブをクリックします。

2. [運用ログ] をクリックし、[実行中のタスクマネージャー] サブタブをクリックして、対応するオペレーターのテクノロジーマネージャー (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
---