This documentation is for an unreleased version of Apache Flink. We recommend you use the latest stable version.

Maxwell Format #

Changelog-Data-Capture Format Format: Serialization Schema Format: Deserialization Schema

Maxwell is a CDC (Changelog Data Capture) tool that can stream changes in real-time from MySQL into Kafka, Kinesis and other streaming connectors. Maxwell provides a unified format schema for changelog and supports to serialize messages using JSON.

Flink supports to interpret Maxwell JSON messages as INSERT/UPDATE/DELETE messages into Flink SQL system. これは次のような多くの場合にこの機能を利用するのに役立ちます

• データベースから他のシステムへの増分データの同期
• 監査ログ
• データベース上でのリアルタイムのマテリアライズドビュー
• データベーステーブルなどの変更履歴の一時結合。

Flink also supports to encode the INSERT/UPDATE/DELETE messages in Flink SQL as Maxwell JSON messages, and emit to external systems like Kafka. ただし、現在FlinkはUPDATE_BEFOREとUPDATE_AFTERを単一のUPDATEメッセージに組み合わせることはできません。Therefore, Flink encodes UPDATE_BEFORE and UDPATE_AFTER as DELETE and INSERT Maxwell messages.

依存 #

In order to use the Maxwell format the following dependencies are required for both projects using a build automation tool (such as Maven or SBT) and SQL Client with SQL JAR bundles.

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-json</artifactId>
  <version>1.19-SNAPSHOT</version>
</dependency>

Note: please refer to Maxwell documentation about how to synchronize changelog to Kafka topics with Maxwell JSON.

How to use Maxwell format #

Maxwell provides a unified format for changelog, here is a simple example for an update operation captured from a MySQL products table in JSON format:

{
   "database":"test",
   "table":"e",
   "type":"insert",
   "ts":1477053217,
   "xid":23396,
   "commit":true,
   "position":"master.000006:800911",
   "server_id":23042,
   "thread_id":108,
   "primary_key": [1, "2016-10-21 05:33:37.523000"],
   "primary_key_columns": ["id", "c"],
   "data":{
     "id":111,
     "name":"scooter",
     "description":"Big 2-wheel scooter",
     "weight":5.15
   },
   "old":{
     "weight":5.18,
   }
}

Note: please refer to Maxwell documentation about the meaning of each fields.

MySQL productsテーブルには4つのカラムがあります(id、name、description、weight)。上記のJSONメッセージは、productsテーブルの更新変更イベントであり、id = 111の行のweight値が、5.18から5.15に変更されます。 Assuming this messages is synchronized to Kafka topic products_binlog, then we can use the following DDL to consume this topic and interpret the change events.

CREATE TABLE topic_products (
  -- schema is totally the same to the MySQL "products" table
  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' = 'maxwell-json'
)

After registering the topic as a Flink table, then you can consume the Maxwell messages as a changelog source.

-- a real-time materialized view on the MySQL "products"
-- which calculate the latest average of weight for the same products
SELECT name, AVG(weight) FROM topic_products GROUP BY name;

-- synchronize all the data and incremental changes of MySQL "products" table to
-- Elasticsearch "products" index for future searching
INSERT INTO elasticsearch_products
SELECT * FROM topic_products;

利用可能なメタデータ #

次のフォーマットのメタデータは、テーブル定義の読み取り専用(VIRTUAL)カラムとして公開できます。

フォーマットメタデータのフィールドは、対応するコネクタがフォーマットメタデータを転送する場合のみ利用できます。現在、Kafkaコネクタだけがその値の形式のメタデータフィールドを公開できます。

The following example shows how to access Maxwell metadata fields in Kafka:

CREATE TABLE KafkaTable (
  origin_database STRING METADATA FROM 'value.database' VIRTUAL,
  origin_table STRING METADATA FROM 'value.table' VIRTUAL,
  origin_primary_key_columns ARRAY<STRING> METADATA FROM 'value.primary-key-columns' VIRTUAL,
  origin_ts TIMESTAMP(3) METADATA FROM 'value.ingestion-timestamp' VIRTUAL,
  user_id BIGINT,
  item_id BIGINT,
  behavior STRING
) WITH (
  'connector' = 'kafka',
  'topic' = 'user_behavior',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'earliest-offset',
  'value.format' = 'maxwell-json'
);

フォーマットオプション #

警告 #

変更イベントの重複 #

The Maxwell application allows to deliver every change event exactly-once. Flink works pretty well when consuming Maxwell produced events in this situation. If Maxwell application works in at-least-once delivery, it may deliver duplicate change events to Kafka and Flink will get the duplicate events. これにより、Flinkクエリで間違った結果や予期しない例外が発生する可能性があります。したがって、上部設定table.exec.source.cdc-events-duplicateをtrueに設定し、この状況ではソースにPRIMARY KEYを定義することをお勧めします。 フレームワークは追加のステートフルオペレーションを生成し、主キーを使って変更イベントの重複排除をし、正規化された変更ログストリームを生成します。

データ型マッピング #

Currently, the Maxwell format uses JSON for serialization and deserialization. Please refer to JSON Format documentation for more details about the data type mapping.