Maxwell

Use cases

• Incremental data sync: Stream database changes to another system in real time.

• Log auditing: Capture and retain a record of every data modification.

• Real-time materialized views: Maintain up-to-date aggregations derived from a source database table.

• Temporal joins: Join a fact stream against a versioned database table using change history.

Example

The following example shows a Maxwell JSON message captured from the products table in a MySQL database. The message records an update to the row where id = 111: the weight field changed from 5.18 to 5.15.

{
   "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
   }
}

For a description of each field, see Maxwell data format.

If this message is published to a Kafka topic named products_binlog, create the following Flink table to consume and parse it:

CREATE TABLE topic_products (
  -- schema is the same as 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'
);

With this table as a changelog source, run queries directly against the MySQL change stream:

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

-- Sync all changes from the MySQL products table to an Elasticsearch index
INSERT INTO elasticsearch_products
SELECT * FROM topic_products;

Parameters

Data type mappings

Maxwell uses JSON for serialization and deserialization. For Flink-to-JSON data type mappings, see JSON format.

Available metadata

The following metadata fields can be declared as read-only VIRTUAL columns in a DDL statement.

The following example declares all four metadata fields in a Kafka table:

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'
);

Limitations