Realtime Compute for Apache Flink:Canal

Use cases

• Incremental data sync: Capture row-level changes from MySQL and propagate them to another system without full table scans.

• Log auditing: Record every INSERT, UPDATE, and DELETE event for compliance and audit trails.

• Real-time materialized views: Keep an aggregated or transformed view of a database table up to date as changes arrive.

• Temporal joins: Enrich streaming data by joining it against a versioned snapshot of a database table.

How it works

When Canal captures a change event from MySQL binlog, it produces a JSON message that includes the changed row data, the source database and table, the change type (INSERT, UPDATE, or DELETE), and timestamps.

Flink reads these messages from Kafka and translates them into Flink SQL changelog events:

• A Canal INSERT event becomes a Flink INSERT message.

• A Canal UPDATE event splits into an UPDATE_BEFORE message (encoded as DELETE) and an UPDATE_AFTER message (encoded as INSERT).

• A Canal DELETE event becomes a Flink DELETE message.

Example

The following Canal JSON message captures an update to the products table in a MySQL database named inventory. The row with id=111 had its weight changed from 5.15 to 5.18.

Key fields in the message:

{
  "data": [
    {
      "id": "111",
      "name": "scooter",
      "description": "Big 2-wheel scooter",
      "weight": "5.18"
    }
  ],
  "database": "inventory",
  "es": 1589373560000,
  "id": 9,
  "isDdl": false,
  "mysqlType": {
    "id": "INTEGER",
    "name": "VARCHAR(255)",
    "description": "VARCHAR(512)",
    "weight": "FLOAT"
  },
  "old": [
    {
      "weight": "5.15"
    }
  ],
  "pkNames": [
    "id"
  ],
  "sql": "",
  "sqlType": {
    "id": 4,
    "name": 12,
    "description": 12,
    "weight": 7
  },
  "table": "products",
  "ts": 1589373560798,
  "type": "UPDATE"
}

For the full Canal JSON field specification, see the Canal documentation.

Assume this message is published to a Kafka topic named products_binlog. Create a Flink table to consume and parse it:

CREATE TABLE topic_products (
  -- Schema must match the products table in MySQL
  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' = 'canal-json'
);

Once the table is defined, run queries against it as a changelog source:

-- Build a real-time materialized view: average weight per product name
SELECT name, AVG(weight) FROM topic_products GROUP BY name;

-- Sync all changes to an Elasticsearch index for search
INSERT INTO elasticsearch_products
SELECT * FROM topic_products;

Parameters

Data type mappings

Canal uses JSON for serialization and deserialization. For data type mappings between Canal JSON and Flink SQL types, see JSON Format.

Canal is also compatible with the Canal extension change type INIT used by Data Transmission Service (DTS) to store data in Kafka clusters. For details, see Data storage formats of Kafka clusters.

Available metadata

The following metadata fields can be declared as read-only VIRTUAL columns in a DDL statement. Metadata fields are only available when the connector propagates format metadata—only the Apache Kafka connector supports this for its value format.

Declare metadata fields with METADATA FROM 'value.<key>' and the VIRTUAL keyword:

CREATE TABLE KafkaTable (
  origin_database STRING METADATA FROM 'value.database' VIRTUAL,
  origin_table STRING METADATA FROM 'value.table' VIRTUAL,
  origin_sql_type MAP<STRING, INT> METADATA FROM 'value.sql-type' VIRTUAL,
  origin_pk_names ARRAY<STRING> METADATA FROM 'value.pk-names' 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' = 'canal-json'
);

Caveats