Realtime Compute for Apache Flink:Upsert Kafka

Prerequisites

Before you begin, ensure that you have:

• A Kafka cluster. Create a DataFlow Kafka cluster or create resources in ApsaraMQ for Kafka

• Network connectivity between your Flink cluster and Kafka cluster. For Kafka on EMR, configure a VPC and security group. For ApsaraMQ for Kafka, configure a whitelist

How it works

The upsert changelog stream produced by the source contains only +I (insert), +U (update after), and -D (delete) entries. Unlike a retract stream, it never contains -U (update before) messages.

As a source table, the connector reads Kafka records and converts them into a changelog stream:

• A record whose key already exists is interpreted as UPDATE (+U)

• A record whose key does not exist is interpreted as INSERT (+I)

• A record with a null value is interpreted as DELETE (-D)

As a sink table or data ingestion sink, the connector consumes an upstream changelog stream:

• INSERT and UPDATE_AFTER records are written as normal Kafka messages

• DELETE records are written as tombstone messages (null value for the corresponding key)

Flink partitions sink output by primary key, so all messages for the same key land in the same partition in the correct order.

Example

The following end-to-end example reads website pageview events from a Kafka source table, aggregates page views (PV) and unique visitors (UV) per region, and writes the results to an Upsert Kafka sink.

Source table — reads raw pageview events:

CREATE TABLE pageviews (
  user_id    BIGINT,
  page_id    BIGINT,
  viewtime   TIMESTAMP,
  user_region STRING,
  WATERMARK FOR viewtime AS viewtime - INTERVAL '2' SECOND
) WITH (
  'connector'                  = 'kafka',
  'topic'                      = '<yourTopicName>',
  'properties.bootstrap.servers' = '<host:port>',
  'format'                     = 'json'
);

Sink table — stores aggregated results with upsert semantics:

CREATE TABLE pageviews_per_region (
  user_region STRING,
  pv          BIGINT,
  uv          BIGINT,
  PRIMARY KEY (user_region) NOT ENFORCED
) WITH (
  'connector'                  = 'upsert-kafka',
  'topic'                      = '<yourTopicName>',
  'properties.bootstrap.servers' = '<host:port>',
  'key.format'                 = 'avro',
  'value.format'               = 'avro'
);

Every Upsert Kafka table must define a primary key. The primary key columns determine how Flink partitions output and how the connector identifies upsert operations.

Query — aggregates and writes results:

INSERT INTO pageviews_per_region
SELECT
  user_region,
  COUNT(*),
  COUNT(DISTINCT user_id)
FROM pageviews
GROUP BY user_region;

SQL

CREATE TABLE upsert_kafka_sink (
  user_region STRING,
  pv          BIGINT,
  uv          BIGINT,
  PRIMARY KEY (user_region) NOT ENFORCED
) WITH (
  'connector'                  = 'upsert-kafka',
  'topic'                      = '<yourTopicName>',
  'properties.bootstrap.servers' = '...',
  'key.format'                 = 'avro',
  'value.format'               = 'avro'
);

Data ingestion

The Upsert Kafka connector can serve as the sink in a YAML data ingestion job. Data is written in JSON format, and primary key fields are included in the message body.

sink:
  type: upsert-kafka
  name: upsert-kafka Sink
  properties.bootstrap.servers: localhost:9092
  # ApsaraMQ for Kafka
  aliyun.kafka.accessKeyId: ${secret_values.kafka-ak}
  aliyun.kafka.accessKeySecret: ${secret_values.kafka-sk}
  aliyun.kafka.instanceId: ${instancd-id}
  aliyun.kafka.endpoint: ${endpoint}
  aliyun.kafka.regionId: ${region-id}

source:
  type: mysql
  name: MySQL Source
  hostname: ${mysql.hostname}
  port: ${mysql.port}
  username: ${mysql.username}
  password: ${mysql.password}
  tables: ${mysql.source.table}
  server-id: 8601-8604

sink:
  type: upsert-kafka
  name: Upsert Kafka Sink
  properties.bootstrap.servers: ${upsert.kafka.bootstraps.server}
  aliyun.kafka.accessKeyId: ${upsert.kafka.aliyun.ak}
  aliyun.kafka.accessKeySecret: ${upsert.kafka.aliyun.sk}
  aliyun.kafka.instanceId: ${upsert.kafka.aliyun.instanceid}
  aliyun.kafka.endpoint: ${upsert.kafka.aliyun.endpoint}
  aliyun.kafka.regionId: ${upsert.kafka.aliyun.regionid}

route:
  - source-table: ${mysql.source.table}
    sink-table: ${upsert.kafka.topic}

Delivery semantics

By default, the Upsert Kafka sink writes with at-least-once guarantees. Flink may write duplicate records with the same key to the topic. Because the connector operates in upsert mode, the last record for any given key takes effect when reading the topic back as a source. Duplicate writes are therefore idempotent — they do not corrupt the final state.

To enable exactly-once semantics, the Kafka cluster must support Kafka transactions (Apache Kafka 0.11 or later) and the transaction feature must be enabled. Set sink.delivery-guarantee to exactly-once in the data ingestion YAML configuration.

Limits

• Requires Flink running on Ververica Runtime (VVR) 2.0.0 or later.

• Supports Apache Kafka 0.10 or later for both reads and writes.

• Supports only the client parameters documented for Apache Kafka 2.8. See the consumer and producer configuration references.

• Upsert Kafka source tables always start from earliest-offset — this is fixed and not configurable. The connector must read all historical change data to reconstruct a complete changelog. Using any other startup mode (such as latest-offset or by timestamp) would result in an incomplete changelog and cause data correctness issues in downstream computations.

• For exactly-once sink semantics, the destination Kafka cluster must run Apache Kafka 0.11 or later with transactions enabled.

Monitoring metrics

Source table metrics

Sink table metrics

What's next

• Synchronize an entire MySQL database to Kafka using Flink CDC

• Resolve Flink dependency conflicts