Apache Flink CDC

https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs
Admin
Apache Flink CDC is a streaming data integration tool that captures real-time changes from databases...
Tokens:103,951
Snippets:1,111
Trust Score:10
Update:3 months ago
Show doc for...
Tokens:
Raw
### Start Docker Compose for PolarDB-X, Elasticsearch, and Kibana

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/connectors/flink-sources/tutorials/polardbx-tutorial

Use this docker-compose.yml to define and start the necessary services for the demo. Ensure Docker is installed.

```yaml
version: '2.1'
services:
  polardbx:
    polardbx:
    image: polardbx/polardb-x:2.0.1
    container_name: polardbx
    ports:
      - "8527:8527"
  elasticsearch:
    image: 'elastic/elasticsearch:7.6.0'
    container_name: elasticsearch
    environment:
      - cluster.name=docker-cluster
      - bootstrap.memory_lock=true
      - ES_JAVA_OPTS=-Xms512m -Xmx512m
      - discovery.type=single-node
    ports:
      - '9200:9200'
      - '9300:9300'
    ulimits:
      memlock:
        soft: -1
        hard: -1
      nofile:
        soft: 65536
        hard: 65536
  kibana:
    image: 'elastic/kibana:7.6.0'
    container_name: kibana
    ports:
      - '5601:5601'
    volumes:
      - '/var/run/docker.sock:/var/run/docker.sock'

```

--------------------------------

### CdcUp Initialization Prompt

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/get-started/quickstart-for-1.20/cdc-up-quickstart-guide

Shows the interactive prompt for initializing the cdc-up playground environment, allowing selection of Flink version. Use this to start the setup process.

```bash
🎉 Welcome to cdc-up quickstart wizard!
   There are a few questions to ask before getting started:
🐿️ Which Flink version would you like to use? (Press ↑/↓ arrow to move and Enter to select)
  1.17.2
  1.18.1
  1.19.3
‣ 1.20.3

```

--------------------------------

### Debezium JSON Example (Initial Data)

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/get-started/quickstart-for-1.20/mysql-to-kafka

Example of Debezium JSON format for initial data synchronization (create operations).

```json
{
    "before": null,
    "after": {
        "id": 1,
        "price": 4
    },
    "op": "c",
    "source": {
        "db": "app_db",
        "table": "orders"
    }
}
// ...
{
    "before": null,
    "after": {
        "id": 1,
        "product": "Beer"
    },
    "op": "c",
    "source": {
        "db": "app_db",
        "table": "products"
    }
}
// ...
{
    "before": null,
    "after": {
        "id": 2,
        "city": "xian"
    },
    "op": "c",
    "source": {
        "db": "app_db",
        "table": "shipments"
    }
}


```

--------------------------------

### Start Docker Containers

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/connectors/flink-sources/tutorials/db2-tutorial

Starts all defined Docker services in detached mode. Use 'docker ps' to verify.

```bash
docker-compose up -d

```

--------------------------------

### DB2 CDC Incremental DataStream Source Example

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/connectors/flink-sources/db2-cdc

Shows how to set up the DB2 CDC connector for incremental change data capture using `StartupOptions.initial()`. This example configures connection details, a deserializer, and specifies startup options for the incremental source.

```java
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.cdc.connectors.base.options.StartupOptions;
import org.apache.flink.cdc.connectors.db2.source.Db2SourceBuilder;
import org.apache.flink.cdc.connectors.db2.source.Db2SourceBuilder.Db2IncrementalSource;
import org.apache.flink.cdc.debezium.JsonDebeziumDeserializationSchema;

public class Db2ParallelSourceExample {

  public static void main(String[] args) throws Exception {

    Db2IncrementalSource<String> sqlServerSource =
            new Db2SourceBuilder()
                    .hostname("localhost")
                    .port(50000)
                    .databaseList("TESTDB")
                    .tableList("DB2INST1.CUSTOMERS")
                    .username("flink")
                    .password("flinkpw")
                    .deserializer(new JsonDebeziumDeserializationSchema())
                    .startupOptions(StartupOptions.initial())
                    .build();

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    // enable checkpoint
    env.enableCheckpointing(3000);
    // set the source parallelism to 2
    env.fromSource(sqlServerSource, WatermarkStrategy.noWatermarks(), "Db2IncrementalSource")
            .setParallelism(2)
            .print()
            .setParallelism(1);

    env.execute("Print DB2 Snapshot + Change Stream");
  }
}

```

--------------------------------

### MySQL CDC DataStream Source Setup

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/connectors/flink-sources/mysql-cdc

Provides a complete Java example for setting up a MySQL CDC DataStream source in Flink. This includes configuring connection details, table discovery, and using a JSON deserializer. Ensure to enable checkpointing for reliable data processing.

```java
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.cdc.connectors.mysql.source.MySqlSource;

public class MySqlSourceExample {
  public static void main(String[] args) throws Exception {
    MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
        .hostname("yourHostname")
        .port(yourPort)
        .databaseList("yourDatabaseName") // set captured database, If you need to synchronize the whole database, Please set tableList to ".*."
        .tableList("yourDatabaseName.yourTableName") // set captured table
        .username("yourUsername")
        .password("yourPassword")
        .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
        .build();

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    // enable checkpoint
    env.enableCheckpointing(3000);

    env
      .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
      // set 4 parallel source tasks
      .setParallelism(4)
      .print().setParallelism(1); // use parallelism 1 for sink to keep message ordering

    env.execute("Print MySQL Snapshot + Binlog");
  }
}
```

--------------------------------

### MySQL to Doris Pipeline Example

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/connectors/pipeline-connectors/mysql

An example pipeline configuration for reading data from MySQL and sinking it to Doris. This snippet demonstrates the source and sink configurations, including table patterns and connection details.

```yaml
source:
   type: mysql
   name: MySQL Source
   hostname: 127.0.0.1
   port: 3306
   username: admin
   password: pass
   tables: adb.\.*, bdb.user_table_[0-9]+, [app|web].order_\.*
   server-id: 5401-5404

sink:
  type: doris
  name: Doris Sink
  fenodes: 127.0.0.1:8030
  username: root
  password: pass

pipeline:
   name: MySQL to Doris Pipeline
   parallelism: 4

```

--------------------------------

### kubectl apply command output

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/deployment/kubernetes

Example output indicating successful creation of the FlinkDeployment resource in Kubernetes.

```text
flinkdeployment.flink.apache.org/flink-cdc-pipeline-job created
```

--------------------------------

### Docker Compose Configuration for Oracle, Elasticsearch, and Kibana

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/connectors/flink-sources/tutorials/oracle-tutorial

Sets up the necessary services for the tutorial. Ensure Docker is installed and running.

```yaml
version: '2.1'
services:
  oracle:
    image: goodboy008/oracle-19.3.0-ee:non-cdb
    ports:
      - "1521:1521"
  elasticsearch:
    image: elastic/elasticsearch:7.6.0
    environment:
      - cluster.name=docker-cluster
      - bootstrap.memory_lock=true
      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
      - discovery.type=single-node
    ports:
      - "9200:9200"
      - "9300:9300"
    ulimits:
      memlock:
        soft: -1
        hard: -1
      nofile:
        soft: 65536
        hard: 65536
  kibana:
    image: elastic/kibana:7.6.0
    ports:
      - "5601:5601"
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock

```

--------------------------------

### Oracle to Doris Pipeline Example

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/connectors/pipeline-connectors/oracle

An example pipeline configuration for reading data from an Oracle source and sinking it to Doris. This demonstrates the basic structure for source, sink, and pipeline definitions.

```yaml
source:
   type: oracle
   name: Oracle Source
   hostname: 127.0.0.1
   port: 1521
   username: debezium
   password: password
   database: ORCLDB
   tables: testdb.\.*, testdb.user_table_[0-9]+, [app|web].order_\.*

sink:
  type: doris
  name: Doris Sink
  fenodes: 127.0.0.1:8030
  username: root
  password: password

pipeline:
   name: Oracle to Doris Pipeline
   parallelism: 4

```

--------------------------------

### Start Flink SQL Client

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/get-started/quickstart-for-1.20/postgres-to-fluss

Command to launch the Flink SQL Client for interacting with data sources.

```bash
bin/sql-client.sh

```

--------------------------------

### Start Docker Compose Services

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/connectors/flink-sources/tutorials/oceanbase-tutorial

Command to start the services defined in the docker-compose.yml file in detached mode.

```bash
docker-compose up -d
```

--------------------------------

### Doris Sink Configuration Example

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/connectors/pipeline-connectors/doris

Example configuration for setting up a Doris sink in a Flink pipeline. It specifies the source type, sink type, connection details for Doris, and pipeline parallelism.

```yaml
source:
   type: values
   name: ValuesSource

sink:
   type: doris
   name: Doris Sink
   fenodes: 127.0.0.1:8030
   username: root
   password: ""
   table.create.properties.replication_num: 1

pipeline:
   parallelism: 1


```

--------------------------------

### Start Flink Cluster

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/connectors/flink-sources/tutorials/build-streaming-etl-tutorial

Start the Flink cluster to enable Flink services. Access the Flink UI at http://localhost:8081.

```bash
./bin/start-cluster.sh
```

--------------------------------

### PostgreSQL CDC Example with Metadata Columns

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/connectors/pipeline-connectors/postgres

This example demonstrates configuring a PostgreSQL CDC source to include metadata columns (op_ts, table_name, schema_name) and projecting them in the transform step.

```yaml
source:
  type: postgres
  hostname: localhost
  port: 5432
  username: postgres
  password: postgres
  tables: mydb.public.orders
  slot.name: flink_slot
  metadata.list: op_ts,table_name,schema_name

transform:
  - source-table: mydb.public.orders
    projection: order_id, customer_id, op_ts, table_name, schema_name
    description: Include metadata columns in output

```

--------------------------------

### Start TiDB Cluster and Test Connection

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/connectors/flink-sources/tutorials/tidb-tutorial

Starts all defined Docker containers in detached mode and then tests the TiDB cluster connection using the MySQL client.

```bash
docker-compose up -d
mysql -h 127.0.0.1 -P 4000 -u root # Just test tidb cluster is ready,if you have install mysql local.
```

--------------------------------

### Product Record Structure

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/get-started/quickstart-for-2.2/mysql-to-kafka

Example of a record pushed to the `yaml-mysql-kafka-products` topic, representing a 'create' operation.

```json
{
"before": null,
"after": {
    "id": 2,
    "product": "Cap"
},
"op": "c",
"source": {
    "db": "app_db",
    "table": "products"
    }
}

```

--------------------------------

### Download and Set Flink Home

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/get-started/quickstart-for-1.20/mysql-to-kafka

Download Flink 1.20.3 and set the FLINK_HOME environment variable. This is the initial setup for the Flink cluster.

```bash
tar -zxvf flink-1.20.3-bin-scala_2.12.tgz
export FLINK_HOME=$(pwd)/flink-1.20.3
cd flink-1.20.3
```

--------------------------------

### Postgres to Fluss Pipeline Example

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/connectors/pipeline-connectors/postgres

An example configuration for a Flink pipeline that reads data from a Postgres source and sinks it to Fluss. Ensure all tables share the same database.

```yaml
source:
   type: postgres
   name: Postgres Source
   hostname: 127.0.0.1
   port: 5432
   username: admin
   password: pass
   # make sure all the tables share same database.
   tables: adb.\.*.\.*
   decoding.plugin.name:  pgoutput
   slot.name: pgtest

sink:
  type: fluss
  name: Fluss Sink
  bootstrap.servers: localhost:9123
  # Security-related properties for the Fluss client
  properties.client.security.protocol: sasl
  properties.client.security.sasl.mechanism: PLAIN
  properties.client.security.sasl.username: developer
  properties.client.security.sasl.password: developer-pass

pipeline:
   name: Postgres to Fluss Pipeline
   parallelism: 4


```

--------------------------------

### MySQL to StarRocks Pipeline Example

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/connectors/pipeline-connectors/starrocks

This example demonstrates a Flink pipeline configuration for reading data from a MySQL source and sinking it to StarRocks. It specifies connection details for both the MySQL source and the StarRocks sink, including JDBC and load URLs.

```yaml
source:
   type: mysql
   name: MySQL Source
   hostname: 127.0.0.1
   port: 3306
   username: admin
   password: pass
   tables: adb.\.*, bdb.user_table_[0-9]+, [app|web].order_\.*
   server-id: 5401-5404

sink:
  type: starrocks
  name: StarRocks Sink
  jdbc-url: jdbc:mysql://127.0.0.1:9030
  load-url: 127.0.0.1:8030
  username: root
  password: pass

pipeline:
   name: MySQL to StarRocks Pipeline
   parallelism: 2

```

--------------------------------

### Start Flink SQL CLI

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/connectors/flink-sources/tutorials/build-streaming-etl-tutorial

Launch the Flink SQL Command Line Interface to interact with Flink SQL.

```bash
./bin/sql-client.sh
```

--------------------------------

### IF Function Example

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/core-concept/transform

Demonstrates the usage of the IF function to return a value based on a simple comparison.

```sql
IF(5 > 3, 5, 3)
```

--------------------------------

### Dockerfile for Flink Environment

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/connectors/flink-sources/tutorials/build-real-time-data-lake-tutorial

Sets up a Flink environment by copying necessary JAR packages into the lib directory and installing the 'tree' utility.

```dockerfile
FROM flink:1.16.0-scala_2.12
# Place the downloaded jar packages in the lib directory at the same level.
COPY ./lib /opt/flink/lib
RUN apt-get update && apt-get install tree

```

--------------------------------

### Navigate to Flink Directory

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/connectors/flink-sources/tutorials/build-streaming-etl-tutorial

Change to the Flink installation directory before running commands.

```bash
cd flink-1.18.0
```

--------------------------------

### Kafka Record Structure for Products

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/get-started/quickstart-for-1.20/mysql-to-kafka

This is an example of a record that will be sent to the `yaml-mysql-kafka-products` topic. It includes before and after states of the row, operation type, and source details.

```json
{
"before": null,
"after": {
    "id": 2,
    "product": "Cap"
},
"op": "c",
"source": {
    "db": "app_db",
    "table": "products"
    }
}
```

--------------------------------

### Prepare MySQL Database and Tables

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/get-started/quickstart-for-1.20/mysql-to-doris

Create the 'app_db' database and 'orders', 'shipments', 'products' tables in MySQL, then insert sample records. Ensure primary keys are defined.

```sql
-- create database
CREATE DATABASE app_db;

USE app_db;

-- create orders table
CREATE TABLE `orders` (
`id` INT NOT NULL,
`price` DECIMAL(10,2) NOT NULL,
PRIMARY KEY (`id`)
);

-- insert records
INSERT INTO `orders` (`id`, `price`) VALUES (1, 4.00);
INSERT INTO `orders` (`id`, `price`) VALUES (2, 100.00);

-- create shipments table
CREATE TABLE `shipments` (
`id` INT NOT NULL,
`city` VARCHAR(255) NOT NULL,
PRIMARY KEY (`id`)
);

-- insert records
INSERT INTO `shipments` (`id`, `city`) VALUES (1, 'beijing');
INSERT INTO `shipments` (`id`, `city`) VALUES (2, 'xian');

-- create products table
CREATE TABLE `products` (
`id` INT NOT NULL,
`product` VARCHAR(255) NOT NULL,
PRIMARY KEY (`id`)
);

-- insert records
INSERT INTO `products` (`id`, `product`) VALUES (1, 'Beer');
INSERT INTO `products` (`id`, `product`) VALUES (2, 'Cap');
INSERT INTO `products` (`id`, `product`) VALUES (3, 'Peanut');

```

--------------------------------

### Create and Switch to 'adb' Database

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/get-started/quickstart-for-1.20/postgres-to-fluss

Create a new database named 'adb' and then switch the current connection to this database for subsequent operations.

```sql
CREATE DATABASE adb;
\c adb

```

--------------------------------

### Create PostgreSQL Database and Schema

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/get-started/quickstart-for-2.2/postgres-to-fluss

Create a new database named 'adb' and switch to it. Then, create 'hr' and 'sales' schemas.

```sql
CREATE DATABASE adb;
\c adb;

-- Create schemas
CREATE SCHEMA hr;
CREATE SCHEMA sales;

```

--------------------------------

### MySQL CDC Source with Regex Table Matching

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/connectors/flink-sources/mysql-cdc

This example shows how to use regular expressions in 'database-name' and 'table-name' options to match multiple tables for the MySQL CDC connector. It demonstrates matching databases starting with 'test', 'tpc', specific names like 'txc', or ending with 'p', and tables like 't5' through 't8' or 'tt'.

```sql
CREATE TABLE products
(
    db_name       STRING METADATA FROM 'database_name' VIRTUAL,
    table_name    STRING METADATA FROM 'table_name' VIRTUAL,
    operation_ts  TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL,
    operation     STRING METADATA FROM 'row_kind' VIRTUAL,
    order_id      INT,
    order_date    TIMESTAMP(0),
    customer_name STRING,
    price         DECIMAL(10, 5),
    product_id    INT,
    order_status  BOOLEAN,
    PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
      'connector' = 'mysql-cdc',
      'hostname' = 'localhost',
      'port' = '3306',
      'username' = 'root',
      'password' = '123456',
      'database-name' = '(^(test).*|^(tpc).*|txc|.*[p$]|t{2})',
      'table-name' = '(t[5-8]|tt)'
      );
```

--------------------------------

### Configure Startup Mode to Latest Offset

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/faq/faq

Set 'scan.startup.mode' to 'latest-offset' to skip the initial snapshot (stock reading phase) and only begin reading from the binlog.

```properties
'scan.startup.mode' = 'latest-offset'.
```

--------------------------------

### Debezium JSON Example (Update Operation)

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/get-started/quickstart-for-1.20/mysql-to-kafka

Example of Debezium JSON format for an update operation on the 'orders' table.

```json
{
    "before": {
        "id": 1,
        "price": 4,
        "amount": null
    },
    "after": {
        "id": 1,
        "price": 100,
        "amount": "100.00"
    },
    "op": "u",
    "source": {
        "db": "app_db",
        "table": "orders"
    }
}


```

--------------------------------

### Create and Populate MySQL Tables (Database 1)

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/connectors/flink-sources/tutorials/build-real-time-data-lake-tutorial

SQL statements to create a 'user' table in 'db_1' and insert initial data.

```sql
 CREATE DATABASE db_1;
 USE db_1;
 CREATE TABLE user_1 (
   id INTEGER NOT NULL PRIMARY KEY,
   name VARCHAR(255) NOT NULL DEFAULT 'flink',
   address VARCHAR(1024),
   phone_number VARCHAR(512),
   email VARCHAR(255)
 );
 INSERT INTO user_1 VALUES (110,"user_110","Shanghai","123567891234","user_110@foo.com");

 CREATE TABLE user_2 (
   id INTEGER NOT NULL PRIMARY KEY,
   name VARCHAR(255) NOT NULL DEFAULT 'flink',
   address VARCHAR(1024),
   phone_number VARCHAR(512),
   email VARCHAR(255)
 );
INSERT INTO user_2 VALUES (120,"user_120","Shanghai","123567891234","user_120@foo.com");

```

--------------------------------

### Start Flink Cluster

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/get-started/quickstart-for-1.20/postgres-to-fluss

Initiate the Flink cluster using the provided script. This command starts the necessary Flink services.

```bash
./bin/start-cluster.sh

```

--------------------------------

### Example Kafka Messages from Different Partitions

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/get-started/quickstart-for-1.20/mysql-to-kafka

Sample output showing messages from partition 0 (an order) and partition 4 (a product and a shipment). This illustrates how data is distributed across partitions when using 'hash-by-key'.

```text
// partition 0
{
    "before": null,
    "after": {
        "id": 1,
        "price": 100,
        "amount": "100.00"
    },
    "op": "c",
    "source": {
        "db": "app_db",
        "table": "orders"
    }
}
// partition 4
{
    "before": null,
    "after": {
        "id": 2,
        "product": "Cap"
    },
    "op": "c",
    "source": {
        "db": "app_db",
        "table": "products"
    }
}
{
    "before": null,
    "after": {
        "id": 1,
        "city": "beijing"
    },
    "op": "c",
    "source": {
        "db": "app_db",
        "table": "shipments"
    }
}

```

--------------------------------

### Create and Populate MySQL Tables (Database 2)

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/connectors/flink-sources/tutorials/build-real-time-data-lake-tutorial

SQL statements to create a 'user' table in 'db_2' and insert initial data, including a NULL email.

```sql
CREATE DATABASE db_2;
USE db_2;
CREATE TABLE user_1 (
  id INTEGER NOT NULL PRIMARY KEY,
  name VARCHAR(255) NOT NULL DEFAULT 'flink',
  address VARCHAR(1024),
  phone_number VARCHAR(512),
  email VARCHAR(255)
);
INSERT INTO user_1 VALUES (110,"user_110","Shanghai","123567891234", NULL);

CREATE TABLE user_2 (
  id INTEGER NOT NULL PRIMARY KEY,
  name VARCHAR(255) NOT NULL DEFAULT 'flink',
  address VARCHAR(1024),
  phone_number VARCHAR(512),
  email VARCHAR(255)
);
INSERT INTO user_2 VALUES (220,"user_220","Shanghai","123567891234","user_220@foo.com");


```

--------------------------------

### Start Local Flink Cluster

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/deployment/standalone

Starts a local Flink cluster using the provided bash script. Navigate to the Flink directory first.

```bash
cd /path/flink-*
./bin/start-cluster.sh

```

--------------------------------

### Debezium-JSON Output Example

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/connectors/pipeline-connectors/kafka

Example of debezium-json output format. This format includes 'before', 'after', 'op', and 'source' elements, but excludes 'ts_ms' from 'source'.

```json
{
  "before": null,
  "after": {
    "col1": "1",
    "col2": "1"
  },
  "op": "c",
  "source": {
    "db": "default_namespace",
    "table": "table1"
  }
}

```

--------------------------------

### Canal-JSON Output Example

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/connectors/pipeline-connectors/kafka

Example of canal-json output format. This format includes 'old', 'data', 'type', 'database', 'table', and 'pkNames' elements, but excludes 'ts'.

```json
{
    "old": null,
    "data": [
        {
            "col1": "1",
            "col2": "1"
        }
    ],
    "type": "INSERT",
    "database": "default_schema",
    "table": "table1",
    "pkNames": [
        "col1"
    ]
}

```

--------------------------------

### String Splitting Column Example

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/connectors/flink-sources/mysql-cdc

Demonstrates how a table with a string primary key is split into chunks using a query to find low and high values for each chunk.

```text
(-∞, 'uuid-001'),
['uuid-001', 'uuid-009'),
['uuid-009', 'uuid-abc'),
['uuid-abc', 'uuid-def'),
[uuid-def, +∞).

```

--------------------------------

### Create MySQL Tables and Data

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/connectors/flink-sources/tutorials/build-streaming-etl-tutorial

SQL statements to create 'products' and 'orders' tables and populate them with initial data in the MySQL database.

```sql
-- MySQL
CREATE DATABASE mydb;
USE mydb;
CREATE TABLE products (
  id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,
  name VARCHAR(255) NOT NULL,
  description VARCHAR(512)
);
ALTER TABLE products AUTO_INCREMENT = 101;

INSERT INTO products
VALUES (default,"scooter","Small 2-wheel scooter"),
       (default,"car battery","12V car battery"),
       (default,"12-pack drill bits","12-pack of drill bits with sizes ranging from #40 to #3"),
       (default,"hammer","12oz carpenter's hammer"),
       (default,"hammer","14oz carpenter's hammer"),
       (default,"hammer","16oz carpenter's hammer"),
       (default,"rocks","box of assorted rocks"),
       (default,"jacket","water resistent black wind breaker"),
       (default,"spare tire","24 inch spare tire");

CREATE TABLE orders (
  order_id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,
  order_date DATETIME NOT NULL,
  customer_name VARCHAR(255) NOT NULL,
  price DECIMAL(10, 5) NOT NULL,
  product_id INTEGER NOT NULL,
  order_status BOOLEAN NOT NULL -- Whether order has been placed
) AUTO_INCREMENT = 10001;

INSERT INTO orders
VALUES (default, '2020-07-30 10:08:22', 'Jark', 50.50, 102, false),
       (default, '2020-07-30 10:11:09', 'Sally', 15.00, 105, false),
       (default, '2020-07-30 12:00:30', 'Edward', 25.25, 106, false);


```

--------------------------------

### Vitess DataStream Source Example

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/connectors/flink-sources/vitess-cdc

Demonstrates how to configure and use the Vitess CDC connector as a DataStream Source in Flink. This example shows setting up connection details and a deserialization schema.

```java
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.cdc.connectors.vitess.VitessSource;

public class VitessSourceExample {
  public static void main(String[] args) throws Exception {
    SourceFunction<String> sourceFunction = VitessSource.<String>builder()
      .hostname("localhost")
      .port(15991)
      .keyspace("inventory")
      .username("flinkuser")
      .password("flinkpw")
      .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
      .build();

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    env
      .addSource(sourceFunction)
      .print().setParallelism(1); // use parallelism 1 for sink to keep message ordering

    env.execute();
  }
}

```

--------------------------------

### Prepare Oracle Database Schema and Data

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/connectors/flink-sources/tutorials/oracle-tutorial

Creates and populates the PRODUCTS and ORDERS tables, enabling supplemental logging for CDC.

```sql
BEGIN
EXECUTE IMMEDIATE 'DROP TABLE DEBEZIUM.PRODUCTS';
EXCEPTION
   WHEN OTHERS THEN
      IF SQLCODE != -942 THEN
         RAISE;
END IF;
END;
/

CREATE TABLE DEBEZIUM.PRODUCTS (
  ID NUMBER(9, 0) NOT NULL,
  NAME VARCHAR(255) NOT NULL,
  DESCRIPTION VARCHAR(512),
  WEIGHT FLOAT,
  PRIMARY KEY(ID)
);

BEGIN
EXECUTE IMMEDIATE 'DROP TABLE DEBEZIUM.ORDERS';
EXCEPTION
   WHEN OTHERS THEN
      IF SQLCODE != -942 THEN
         RAISE;
END IF;
END;
/

CREATE TABLE DEBEZIUM.ORDERS (
  ID NUMBER(9, 0) NOT NULL,
  ORDER_DATE TIMESTAMP(3) NOT NULL,
  PURCHASER VARCHAR(255) NOT NULL,
  QUANTITY NUMBER(9, 0) NOT NULL,
  PRODUCT_ID NUMBER(9, 0) NOT NULL,
  PRIMARY KEY(ID)
);

ALTER TABLE DEBEZIUM.PRODUCTS ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
ALTER TABLE DEBEZIUM.ORDERS ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;

INSERT INTO DEBEZIUM.PRODUCTS VALUES (101, 'scooter', 'Small 2-wheel scooter', 3.14);
INSERT INTO DEBEZIUM.PRODUCTS VALUES (102, 'car battery', '12V car battery', 8.1);
INSERT INTO DEBEZIUM.PRODUCTS VALUES (103, '12-pack drill bits', '12-pack of drill bits with sizes ranging from #40 to #3', 0.8);
INSERT INTO DEBEZIUM.PRODUCTS VALUES (104, 'hammer', '12oz carpenter''s hammer', 0.75);
INSERT INTO DEBEZIUM.PRODUCTS VALUES (105, 'hammer', '14oz carpenter''s hammer', 0.875);
INSERT INTO DEBEZIUM.PRODUCTS VALUES (106, 'hammer', '16oz carpenter''s hammer', 1.0);
INSERT INTO DEBEZIUM.PRODUCTS VALUES (107, 'rocks', 'box of assorted rocks', 5.3);
INSERT INTO DEBEZIUM.PRODUCTS VALUES (108, 'jacket', 'water resistent black wind breaker', 0.1);
INSERT INTO DEBEZIUM.PRODUCTS VALUES (109, 'spare tire', '24 inch spare tire', 22.2);

INSERT INTO DEBEZIUM.ORDERS VALUES (1001, TO_TIMESTAMP('2020-07-30 10:08:22.001000', 'YYYY-MM-DD HH24:MI:SS.FF'), 'Jark', 1, 101);
INSERT INTO DEBEZIUM.ORDERS VALUES (1002, TO_TIMESTAMP('2020-07-30 10:11:09.001000', 'YYYY-MM-DD HH24:MI:SS.FF'), 'Sally', 2, 102);
INSERT INTO DEBEZIUM.ORDERS VALUES (1003, TO_TIMESTAMP('2020-07-30 12:00:30.001000', 'YYYY-MM-DD HH24:MI:SS.FF'), 'Edward', 2, 103);
INSERT INTO DEBEZIUM.ORDERS VALUES (1004, TO_TIMESTAMP('2020-07-30 15:22:00.001000', 'YYYY-MM-DD HH24:MI:SS.FF'), 'Jark', 1, 104);

```

--------------------------------

### Canal JSON Output Format Example

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/get-started/quickstart-for-1.20/mysql-to-kafka

Example of a Canal JSON message representing an INSERT operation. It includes 'old' (null for insert), 'data', 'type', 'database', 'table', and 'pkNames'.

```json
{
    "old": null,
    "data": [
        {
            "id": 1,
            "price": 100,
            "amount": "100.00"
        }
    ],
    "type": "INSERT",
    "database": "app_db",
    "table": "orders",
    "pkNames": [
        "id"
    ]
}

```

--------------------------------

### Connect to PostgreSQL

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/get-started/quickstart-for-1.20/postgres-to-fluss

Establish a connection to the PostgreSQL database using the psql client. Ensure you have the correct host, port, user, and database name.

```bash
psql -h localhost -p 5432 -U root postgres

```

--------------------------------

### Debezium JSON Output Example

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/get-started/quickstart-for-1.20/mysql-to-kafka

Example of a Debezium JSON message representing a record insertion ('c' operation) with 'before' and 'after' states. The 'source.table' field indicates the original table name.

```json
{
    "before": null,
    "after": {
        "id": 1,
        "price": 100,
        "amount": "100.00"
    },
    "op": "c",
    "source": {
        "db": null,
        "table": "kafka_ods_orders"
    }
}

```

--------------------------------

### MySQL Initialization and Test Data

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/get-started/quickstart-for-1.20/cdc-up-quickstart-guide

SQL commands to create a database, a table, insert test data, and verify the insertion. Use this when MySQL is chosen as the data source to prepare the environment.

```sql
-- initialize db and table
CREATE DATABASE cdc_playground;
USE cdc_playground;
CREATE TABLE test_table (id INT PRIMARY KEY, name VARCHAR(32));

-- insert test data
INSERT INTO test_table VALUES (1, 'alice'), (2, 'bob'), (3, 'cicada'), (4, 'derrida');

-- verify if it has been successfully inserted
SELECT * FROM test_table;

```

--------------------------------

### MySQL CDC Startup Options in DataStream API

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/connectors/flink-sources/mysql-cdc

Demonstrates various startup modes for the MySQL CDC connector using the DataStream API builder. Choose the appropriate option based on your needs, such as starting from the earliest, latest, a specific offset, a timestamp, or only performing a snapshot.

```java
MySQLSource.builder()
    .startupOptions(StartupOptions.earliest()) // Start from earliest offset
    .startupOptions(StartupOptions.latest()) // Start from latest offset
    .startupOptions(StartupOptions.specificOffset("mysql-bin.000003", 4L)) // Start from binlog file and offset
    .startupOptions(StartupOptions.specificOffset("24DA167-0C0C-11E8-8442-00059A3C7B00:1-19")) // Start from GTID set
    .startupOptions(StartupOptions.timestamp(1667232000000L)) // Start from timestamp
    .startupOptions(StartupOptions.snapshot()) // Read snapshot only
    ...
    .build()
```

--------------------------------

### MongoDB CDC Connector Startup Options (DataStream API)

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/connectors/flink-sources/mongodb-cdc

Configures the startup mode for the MongoDB CDC consumer in the Flink DataStream API. Use `StartupOptions.latest()` to start from the latest offset or `StartupOptions.timestamp()` to start from a specific point in time.

```java
MongoDBSource.builder()
    .startupOptions(StartupOptions.latest()) // Start from latest offset
    .startupOptions(StartupOptions.timestamp(1667232000000L)) // Start from timestamp
    .build()
```

--------------------------------

### Enter MySQL Container

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/get-started/quickstart-for-1.20/mysql-to-doris

Access the MySQL container using the provided credentials to set up the database and tables.

```bash
docker-compose exec mysql mysql -uroot -p123456
```

--------------------------------

### TiDB CDC DataStream Source Example

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/connectors/flink-sources/tidb-cdc

Demonstrates how to create and configure a TiDB CDC source as a Flink DataStream. This example shows setting up database and table capture, TiKV configuration, and custom deserializers for snapshot and change events.

```java
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;

import org.apache.flink.cdc.connectors.tidb.TDBSourceOptions;
import org.apache.flink.cdc.connectors.tidb.TiDBSource;
import org.apache.flink.cdc.connectors.tidb.TiKVChangeEventDeserializationSchema;
import org.apache.flink.cdc.connectors.tidb.TiKVSnapshotEventDeserializationSchema;
import org.tikv.kvproto.Cdcpb;
import org.tikv.kvproto.Kvrpcpb;

import java.util.HashMap;

public class TiDBSourceExample {

    public static void main(String[] args) throws Exception {

        SourceFunction<String> tidbSource =
            TiDBSource.<String>builder()
                .database("mydb") // set captured database
                .tableName("products") // set captured table
                .tiConf(
                    TDBSourceOptions.getTiConfiguration(
                        "localhost:2399", new HashMap<>()))
                .snapshotEventDeserializer(
                    new TiKVSnapshotEventDeserializationSchema<String>() {
                        @Override
                        public void deserialize(
                            Kvrpcpb.KvPair record, Collector<String> out)
                            throws Exception {
                            out.collect(record.toString());
                        }

                        @Override
                        public TypeInformation<String> getProducedType() {
                            return BasicTypeInfo.STRING_TYPE_INFO;
                        }
                    })
                .changeEventDeserializer(
                    new TiKVChangeEventDeserializationSchema<String>() {
                        @Override
                        public void deserialize(
                            Cdcpb.Event.Row record, Collector<String> out)
                            throws Exception {
                            out.collect(record.toString());
                        }

                        @Override
                        public TypeInformation<String> getProducedType() {
                            return BasicTypeInfo.STRING_TYPE_INFO;
                        }
                    })
                .build();

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // enable checkpoint
        env.enableCheckpointing(3000);
        env.addSource(tidbSource).print().setParallelism(1);

        env.execute("Print TiDB Snapshot + Binlog");
    }
}

```

--------------------------------

### Connect to Oracle Database

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/connectors/flink-sources/tutorials/oracle-tutorial

Connects to the Oracle database using SQL*Plus to prepare data. Ensure the Oracle container is running.

```bash
docker-compose exec oracle sqlplus debezium/dbz@localhost:1521/ORCLCDB

```

--------------------------------

### Flink DataStream API MySQL CDC Source Example

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/connectors/flink-sources/overview

Example demonstrating how to configure and use the MySQL CDC source with Flink's DataStream API. This includes setting up connection details, database/table lists, deserialization, and enabling checkpointing.

```java
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.cdc.connectors.mysql.source.MySqlSource;

public class MySqlBinlogSourceExample {
  public static void main(String[] args) throws Exception {
    MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
            .hostname("yourHostname")
            .port(yourPort)
            .databaseList("yourDatabaseName") // set captured database
            .tableList("yourDatabaseName.yourTableName") // set captured table
            .username("yourUsername")
            .password("yourPassword")
            .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
            .build();
    
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
    // enable checkpoint
    env.enableCheckpointing(3000);
    
    env
      .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
      // set 4 parallel source tasks
      .setParallelism(4)
      .print().setParallelism(1); // use parallelism 1 for sink to keep message ordering
    
    env.execute("Print MySQL Snapshot + Binlog");
  }
}
```

--------------------------------

### Docker Compose for MongoDB, Elasticsearch, and Kibana

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/connectors/flink-sources/tutorials/mongodb-tutorial

Sets up the necessary services for the tutorial. Ensure MongoDB is configured with replica set support.

```yaml
version: '2.1'
services:
  mongo:
    image: "mongo:4.0-xenial"
    command: --replSet rs0 --smallfiles --oplogSize 128
    ports:
      - "27017:27017"
    environment:
      - MONGO_INITDB_ROOT_USERNAME=mongouser
      - MONGO_INITDB_ROOT_PASSWORD=mongopw
  elasticsearch:
    image: elastic/elasticsearch:7.6.0
    environment:
      - cluster.name=docker-cluster
      - bootstrap.memory_lock=true
      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
      - discovery.type=single-node
    ports:
      - "9200:9200"
      - "9300:9300"
    ulimits:
      memlock:
        soft: -1
        hard: -1
      nofile:
        soft: 65536
        hard: 65536
  kibana:
    image: elastic/kibana:7.6.0
    ports:
      - "5601:5601"

```

--------------------------------

### Shipment Record Structure

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/get-started/quickstart-for-2.2/mysql-to-kafka

Example of a record pushed to the `yaml-mysql-kafka-shipments` topic, representing a 'create' operation.

```json
{
"before": null,
"after": {
    "id": 2,
    "city": "xian"
},
"op": "c",
"source": {
    "db": "app_db",
    "table": "shipments"
    }
}

```

--------------------------------

### Connect to PostgreSQL

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/get-started/quickstart-for-2.2/postgres-to-fluss

Establish a connection to the PostgreSQL database using the psql client. The password is 'password'.

```bash
psql -h localhost -p 5432 -U root postgres
```

--------------------------------

### Connect to OceanBase Observer

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/connectors/flink-sources/tutorials/oceanbase-tutorial

Connects to the OceanBase observer using obclient as the 'root' user of the 'sys' tenant. Ensure the observer is running and accessible.

```bash
docker-compose exec observer obclient -h127.0.0.1 -P2881 -uroot@sys -p123456
```

--------------------------------

### Order Record Structure

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/get-started/quickstart-for-2.2/mysql-to-kafka

Example of a record pushed to the `yaml-mysql-kafka-orders` topic, representing a 'create' operation.

```json
{
    "before": null,
    "after": {
        "id": 1,
        "price": 100,
        "amount": "100.00"
    },
    "op": "c",
    "source": {
        "db": "app_db",
        "table": "orders"
    }
}

```

--------------------------------

### MySQL to MaxCompute Pipeline Configuration

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/connectors/pipeline-connectors/maxcompute

Example pipeline definition for reading data from MySQL and sinking it to MaxCompute. Ensure all required sink parameters like access-id, access-key, endpoint, and project are correctly configured.

```yaml
source:
  type: mysql
  name: MySQL Source
  hostname: 127.0.0.1
  port: 3306
  username: admin
  password: pass
  tables: adb.\.*, bdb.user_table_[0-9]+, [app|web].order_\.*
  server-id: 5401-5404

sink:
  type: maxcompute
  name: MaxCompute Sink
  access-id: ak
  access-key: sk
  endpoint: endpoint
  project: flink_cdc
  buckets-num: 8

pipeline:
  name: MySQL to MaxCompute Pipeline
  parallelism: 2

```

--------------------------------

### Set Flink Home Environment Variable

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/deployment/standalone

Sets the FLINK_HOME environment variable to the Flink installation directory.

```bash
export FLINK_HOME=/path/flink-*

```

--------------------------------

### Start Flink Session Cluster on Kubernetes

Source: https://nightlies.apache.org/flink/flink-cdc-docs-release-3.6/docs/deployment/kubernetes

Initiate a Flink session cluster on Kubernetes with a specified cluster ID.

```bash
cd /path/flink-*
./bin/kubernetes-session.sh -Dkubernetes.cluster-id=my-first-flink-cluster

```
Tokens:
Raw