Realtime Compute for Apache Flink:Debezium

このトピックでは、Debezium 形式の使用例を示し、形式オプションとデータ型マッピングについて説明します。

背景情報

Debezium は、MySQL、PostgreSQL、Oracle、Microsoft SQL Server などのさまざまなデータベースから Kafka にリアルタイムでデータ変更をストリーミングできる Change Data Capture (CDC) ツールです。Debezium は、変更ログの統一形式スキーマを提供し、JSON および Apache Avro を使用してメッセージをシリアル化できます。Kafka コネクタと オブジェクトストレージサービス (OSS) は、Debezium 形式をサポートしています。

Flink は、Debezium JSON メッセージと Debezium Avro メッセージを、Flink SQL システムの INSERT、UPDATE、または DELETE メッセージとして解釈できます。Debezium 形式は、以下のシナリオに適しています。

• データベースから別のシステムへの増分データの同期

• ログ監査

• データベースのリアルタイム マテリアライズド ビュー

• データベース テーブルの時間的結合

Flink は、Flink SQL システムの INSERT、UPDATE、または DELETE メッセージを Debezium JSON メッセージまたは Debezium Avro メッセージとして解釈し、それらのメッセージを Kafka などのデータストアに転送することもできます。

例

Flink は、MySQL データベースの products テーブルの更新レコードを受信します。レコードは Debezium JSON 形式であり、id、name、description、および weight 列が含まれています。

{
  "before": {
    "id": 111,
    "name": "scooter",
    "description": "Big 2-wheel scooter",  // 大型 2 輪スクーター
    "weight": 5.18
  },
  "after": {
    "id": 111,
    "name": "scooter",
    "description": "Big 2-wheel scooter",  // 大型 2 輪スクーター
    "weight": 5.15
  },
  "source": {...},
  "op": "u",
  "ts_ms": 1589362330904,
  "transaction": null
};

上記の例の JSON メッセージには、products テーブルの更新イベントが含まれています。id の値が 111 の行で、weight の値が 5.18 から 5.15 に変更されています。メッセージが products_binlog という名前の Kafka トピックに同期されている場合は、次の DDL ステートメントを実行してこのトピックを使用し、変更イベントを解釈できます。

-- debezium-json 形式を使用して、Debezium JSON メッセージを解釈します。
CREATE TABLE topic_products (
-- topic_products テーブルのスキーマが、MySQL データベースの products テーブルのスキーマと同じであることを確認します。
  id BIGINT,
  name STRING,
  description STRING,
  weight DECIMAL(10, 2)
) WITH (
 'connector' = 'kafka',
 'topic' = 'products_binlog',
 'properties.bootstrap.servers' = 'localhost:9092',
 'properties.group.id' = 'testGroup',
 'format' = 'debezium-json'
-- debezium-json 形式を使用して、Debezium JSON メッセージを解釈します。
-- Debezium メッセージが Avro メッセージの場合は、debezium-avro-confluent 形式を使用します。
);

Debezium Kafka Connect を構成する場合、ビジネス要件に基づいて value.converter.schemas.enable オプションを true に設定できます。このようにして、スキーマ情報がメッセージ本文に含まれます。次のサンプル コードは、Debezium JSON メッセージの例を示しています。

{
  "schema": {...},  // スキーマ
  "payload": {  // ペイロード
    "before": {
      "id": 111,
      "name": "scooter",  // スクーター
      "description": "Big 2-wheel scooter",  // 大型 2 輪スクーター
      "weight": 5.18
    },
    "after": {
      "id": 111,
      "name": "scooter",  // スクーター
      "description": "Big 2-wheel scooter",  // 大型 2 輪スクーター
      "weight": 5.15
    },
    "source": {...},  // ソース
    "op": "u",  // 操作
    "ts_ms": 1589362330904,  // タイムスタンプ (ミリ秒)
    "transaction": null  // トランザクション
  }
}

Debezium JSON メッセージを解釈するには、'debezium-json.schema-include' = 'true' 構成を上記の例の DDL ステートメントの WITH 句に追加する必要があります。デフォルトでは、debezium-json.schema-include オプションは false に設定されています。ほとんどの場合、スキーマの説明を Debezium JSON メッセージに含めないことをお勧めします。これは、メッセージが長くなり、解析パフォーマンスが低下するためです。

トピックを Flink テーブルとして登録した後、Debezium メッセージを changelog ソースとして使用できます。

-- MySQL データベースの products テーブルのリアルタイム マテリアライズド ビューで、同じ製品の最新の平均重量を計算します。
SELECT name, AVG(weight) FROM topic_products GROUP BY name;

-- MySQL データベースの products テーブルのすべてのデータと増分変更を、後続のクエリのために Elasticsearch の products インデックスに同期します。
INSERT INTO elasticsearch_products
SELECT * FROM topic_products;

形式オプション

Flink は、Debezium Avro メッセージまたは Debezium JSON メッセージを解釈するための debezium-avro-confluent 形式と debezium-json 形式を提供します。

debezium-avro-confluent

Debezium Avro メッセージを解釈するには、debezium-avro-confluent 形式を使用します。

debezium-json

Debezium JSON メッセージを解釈するには、debezium-json 形式を使用します。

データ型マッピング

Debezium は、シリアル化と逆シリアル化に JSON 形式を使用します。データ型マッピングの詳細については、「JSON 形式」のドキュメントと「Confluent Avro 形式」のドキュメントをご参照ください。

その他

使用可能なメタデータ

次の表で説明されている形式メタデータ フィールドは、DDL ステートメントで読み取り専用 (VIRTUAL) 列として宣言できます。

次のサンプル コードは、Kafka の Debezium メタデータ フィールドにアクセスする方法の例を示しています。

CREATE TABLE KafkaTable (
  origin_ts TIMESTAMP(3) METADATA FROM 'value.ingestion-timestamp' VIRTUAL,  // 処理タイムスタンプ
  event_time TIMESTAMP(3) METADATA FROM 'value.source.timestamp' VIRTUAL,  // イベントタイム
  origin_database STRING METADATA FROM 'value.source.database' VIRTUAL,  // データベース名
  origin_schema STRING METADATA FROM 'value.source.schema' VIRTUAL,  // スキーマ名
  origin_table STRING METADATA FROM 'value.source.table' VIRTUAL,  // テーブル名
  origin_properties MAP<STRING, STRING> METADATA FROM 'value.source.properties' VIRTUAL,  // プロパティ
  user_id BIGINT,  // ユーザーID
  item_id BIGINT,  // アイテムID
  behavior STRING  // 動作
) WITH (
  'connector' = 'kafka',
  'topic' = 'user_behavior',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'earliest-offset',
  'value.format' = 'debezium-json'
);

FAQ

障害発生時に重複する変更イベントが配信された場合はどうすればよいですか?

通常の動作環境では、Debezium は exactly-once セマンティクスを使用して、ソース テーブルの各変更イベントを配信できます。その後、Realtime Compute for Apache Flink は、Debezium によって生成された変更イベントを想定どおりに消費できます。障害が発生した場合、Debezium は at-least-once 配信のみを保証できます。この場合、Debezium は重複する変更イベントを Kafka に配信する可能性があります。Realtime Compute for Apache Flink は、Kafka からデータを取得するときに重複イベントを取得します。これは、Realtime Compute for Apache Flink クエリの誤った結果または予期しない例外を引き起こす可能性があります。この問題を回避するには、デプロイ パラメータ table.exec.source.cdc-events-duplicate を true に設定し、ソースに主キーを定義することをお勧めします。このようにして、Realtime Compute for Apache Flink は追加のステートフル演算子を生成できます。主キーは、変更イベントの重複を排除し、正規化された changelog ストリームを生成するために使用されます。

PostgreSQL 用 Debezium コネクタを使用して生成されたデータが想定どおりに解釈されない場合はどうすればよいですか?

PostgreSQL 用 Debezium コネクタを使用して Kafka への変更をキャプチャする場合は、監視対象テーブルの REPLICA IDENTITY が FULL に設定されていることを確認してください。デフォルト値は DEFAULT です。REPLICA IDENTITY が FULL に設定されていない場合、Flink SQL は Debezium データを想定どおりに解釈できません。

REPLICA IDENTITY が FULL に設定されている場合、UPDATE イベントと DELETE イベントには、すべての列の以前の値が含まれます。REPLICA IDENTITY が FULL に設定されていない場合、UPDATE イベントと DELETE イベントの before フィールドには、主キー フィールドの値のみが含まれます。REPLICA IDENTITY が FULL に設定されておらず、主キーが指定されていない場合、UPDATE イベントと DELETE イベントの before フィールドの値は null です。ALTER TABLE <your-table-name> REPLICA IDENTITY FULL ステートメントを実行して、REPLICA IDENTITY の構成を変更できます。