lvyanquan commented on code in PR #4604: URL: https://github.com/apache/flink-cdc/pull/4604#discussion_r4396669521
########## docs/content/docs/examples-and-best-practices/mysql-to-paimon.md: ########## @@ -0,0 +1,301 @@ +--- +title: "Real-time MySQL Ingestion into Paimon" +slug: "/docs/examples-and-best-practices/mysql-to-paimon" +sidebar_position: 2 +hide_title: true +displayed_sidebar: docs +--- +<!-- +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + +http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +KIND, either express or implied. See the License for the +specific language governing permissions and limitations +under the License. +--> + +# Real-time MySQL Ingestion into Paimon + +This example uses a Flink CDC YAML pipeline to merge two MySQL order tables into one Apache Paimon table. +It covers the initial snapshot, schema evolution, and subsequent inserts, updates, and deletes. +SQL queries are used to verify the ingested data. + +The two source tables have the same schema, and their order keys are unique across both tables. +When merging tables with overlapping primary key values, define a composite key that also identifies the source table; +otherwise, records with the same key can overwrite each other in Paimon. + +## Prepare the Flink environment + +Prepare a Linux or macOS machine with Docker Compose, Java 11, and a Flink standalone cluster. +Choose a Flink distribution and the corresponding dependencies from the following table: + +| Flink distribution | Required dependencies | +|--------------------|-----------------------| +| [Flink 1.20.5](https://archive.apache.org/dist/flink/flink-1.20.5/flink-1.20.5-bin-scala_2.12.tgz) | [Flink CDC @@VERSION@@](https://archive.apache.org/dist/flink/flink-cdc-@@VERSION@@/flink-cdc-@@VERSION@@-1.20-bin.tar.gz)<br />[MySQL Pipeline connector](https://repo.maven.apache.org/maven2/org/apache/flink/flink-cdc-pipeline-connector-mysql/@@VERSION@@-1.20/flink-cdc-pipeline-connector-mysql-@@VERSION@@-1.20.jar)<br />[Paimon Pipeline connector](https://repo.maven.apache.org/maven2/org/apache/flink/flink-cdc-pipeline-connector-paimon/@@VERSION@@-1.20/flink-cdc-pipeline-connector-paimon-@@VERSION@@-1.20.jar)<br />[MySQL JDBC 8.0.27](https://repo.maven.apache.org/maven2/mysql/mysql-connector-java/8.0.27/mysql-connector-java-8.0.27.jar)<br />[Hadoop 2.8.3-10.0](https://repo.maven.apache.org/maven2/org/apache/flink/flink-shaded-hadoop-2-uber/2.8.3-10.0/flink-shaded-hadoop-2-uber-2.8.3-10.0.jar)<br />[Paimon SQL 2.1.0 for Flink 1.20](https://repo.maven.apache.org/maven2/org/apache/pa imon/paimon-flink-1.20/2.1.0/paimon-flink-1.20-2.1.0.jar) | +| [Flink 2.2.1](https://archive.apache.org/dist/flink/flink-2.2.1/flink-2.2.1-bin-scala_2.12.tgz) | [Flink CDC @@VERSION@@](https://archive.apache.org/dist/flink/flink-cdc-@@VERSION@@/flink-cdc-@@VERSION@@-2.2-bin.tar.gz)<br />[MySQL Pipeline connector](https://repo.maven.apache.org/maven2/org/apache/flink/flink-cdc-pipeline-connector-mysql/@@VERSION@@-2.2/flink-cdc-pipeline-connector-mysql-@@VERSION@@-2.2.jar)<br />[Paimon Pipeline connector](https://repo.maven.apache.org/maven2/org/apache/flink/flink-cdc-pipeline-connector-paimon/@@VERSION@@-2.2/flink-cdc-pipeline-connector-paimon-@@VERSION@@-2.2.jar)<br />[MySQL JDBC 8.0.27](https://repo.maven.apache.org/maven2/mysql/mysql-connector-java/8.0.27/mysql-connector-java-8.0.27.jar)<br />[Hadoop 2.8.3-10.0](https://repo.maven.apache.org/maven2/org/apache/flink/flink-shaded-hadoop-2-uber/2.8.3-10.0/flink-shaded-hadoop-2-uber-2.8.3-10.0.jar)<br />[Paimon SQL 2.1.0 for Flink 2.1](https://repo.maven.apache.org/maven2/org/apache/paimon/paim on-flink-2.1/2.1.0/paimon-flink-2.1-2.1.0.jar) | + +The Paimon SQL connector in each row matches the Paimon version bundled with the corresponding Pipeline connector. +Download and extract the selected distributions, then set the following paths to your installation directories: + +```shell +export FLINK_HOME=/path/to/flink +export FLINK_CDC_HOME=/path/to/flink-cdc +``` + +Place the MySQL and Paimon Pipeline connector JARs from the table in `$FLINK_CDC_HOME/lib`. + +Place the MySQL JDBC driver from the table in `$FLINK_HOME/lib` before starting Flink. +Save the Hadoop JAR as `$FLINK_HOME/opt/flink-shaded-hadoop.jar` and the Paimon SQL connector +as `$FLINK_HOME/opt/paimon-sql-connector.jar`. +The commands below add Hadoop with `--jar` when submitting the pipeline and verification queries. +If this Hadoop JAR is already in `$FLINK_HOME/lib`, move it to the `opt` path above before starting the cluster. +Loading it as a job dependency avoids conflicts with the libraries used by Flink's own CLI. + +In `$FLINK_HOME/conf/config.yaml`, add or update the following settings. +Merge them into existing `execution` and `taskmanager` sections if those sections already exist: + +```yaml +execution: + checkpointing: + interval: 3s +taskmanager: + numberOfTaskSlots: 4 +``` + +Checkpointing makes Paimon commits visible to readers. +Four task slots allow the ingestion pipeline and verification queries to run together. +Start the cluster: + +```shell +"$FLINK_HOME/bin/start-cluster.sh" +``` + +The [Flink Web UI](http://localhost:8081/) should show an available TaskManager. + +## Prepare the MySQL tables + +Create a working directory and save the following content as `compose.yaml`: + +```yaml +services: + mysql: + image: mysql:8.0 + platform: linux/amd64 + ports: + - "3306:3306" + environment: + MYSQL_ROOT_PASSWORD: tutorial + MYSQL_ROOT_HOST: "%" + command: + - --server-id=1 + - --log-bin=mysql-bin + - --binlog-format=ROW + - --binlog-row-image=FULL + - --default-time-zone=+00:00 + healthcheck: + test: ["CMD", "mysqladmin", "ping", "-h", "127.0.0.1", "-uroot", "-ptutorial"] + interval: 2s + timeout: 5s + retries: 30 +``` + +Start MySQL and wait for its health check to succeed: + +```shell +docker compose up -d --wait +docker compose exec mysql mysql -uroot -ptutorial +``` + +In the MySQL client, create the two source tables and insert sample data: + +```sql +CREATE DATABASE orders; +USE orders; + +CREATE TABLE orders_1 ( + orderkey BIGINT NOT NULL, + custkey BIGINT, + total_price DECIMAL(10, 2), + order_date DATE, + PRIMARY KEY (orderkey) +); +CREATE TABLE orders_2 LIKE orders_1; + +INSERT INTO orders_1 VALUES + (1, 1, 100.00, '2026-01-01'), + (2, 2, 200.00, '2026-01-02'); +INSERT INTO orders_2 VALUES + (11, 1, 300.00, '2026-01-03'), + (12, 3, 400.00, '2026-01-04'); +``` + +Keep this client open for the schema and data changes below. Run the remaining shell commands in another terminal. + +## Define and submit the ingestion pipeline + +Save the following configuration as `mysql-to-paimon.yaml`: + +```yaml +source: + type: mysql + hostname: localhost + port: 3306 + username: root + password: tutorial + tables: orders.orders_[0-9]+ + server-id: 5400-5404 + server-time-zone: UTC + +sink: + type: paimon + catalog.properties.metastore: filesystem + catalog.properties.warehouse: file:///tmp/flink-cdc-paimon-tutorial/warehouse + +route: + - source-table: orders.orders_[0-9]+ + sink-table: ods.orders + +pipeline: + name: MySQL orders to Paimon + parallelism: 2 + schema.change.behavior: evolve +``` + +The `tables` expression selects both order tables, and the route merges them into `ods.orders`. +Flink CDC creates the target database and table automatically. + +The `file:///` warehouse is suitable for this example because all Flink processes run on one machine. +For a distributed cluster, use shared storage accessible from the client, JobManager, and every TaskManager. + +Submit the pipeline: + +```shell +"$FLINK_CDC_HOME/bin/flink-cdc.sh" mysql-to-paimon.yaml \ + --jar "$FLINK_HOME/opt/flink-shaded-hadoop.jar" +``` + +Record the returned Job ID. In the Flink Web UI, confirm that the job is running and a checkpoint completes. + +## Verify the initial snapshot + +Open the Flink SQL Client: + +```shell +"$FLINK_HOME/bin/sql-client.sh" \ + --jar "$FLINK_HOME/opt/paimon-sql-connector.jar" \ + --jar "$FLINK_HOME/opt/flink-shaded-hadoop.jar" +``` + +Create a Paimon catalog pointing to the same warehouse, then query the merged table: + +```sql +SET 'execution.runtime-mode' = 'batch'; +SET 'sql-client.execution.result-mode' = 'tableau'; +SET 'parallelism.default' = '1'; + +CREATE CATALOG lake WITH ( + 'type' = 'paimon', + 'metastore' = 'filesystem', + 'warehouse' = 'file:///tmp/flink-cdc-paimon-tutorial/warehouse', + 'cache-enabled' = 'false' +); +USE CATALOG lake; + +SELECT orderkey, custkey, total_price, order_date +FROM ods.orders +ORDER BY orderkey; +``` + +The result should contain these four rows: + +| orderkey | custkey | total_price | order_date | +|----------|---------|-------------|------------| +| 1 | 1 | 100.00 | 2026-01-01 | +| 2 | 2 | 200.00 | 2026-01-02 | +| 11 | 1 | 300.00 | 2026-01-03 | +| 12 | 3 | 400.00 | 2026-01-04 | + +If the table is not available yet, wait for the snapshot and a successful checkpoint, then run the query again. +Batch queries read a committed snapshot; run them again to see later changes. + +## Synchronize schema changes + +In the MySQL client, add the same column to both source tables and update two records: + +```sql +ALTER TABLE orders.orders_1 ADD COLUMN quantity BIGINT; +ALTER TABLE orders.orders_2 ADD COLUMN quantity BIGINT; +UPDATE orders.orders_1 SET quantity = 2 WHERE orderkey = 1; +UPDATE orders.orders_2 SET quantity = 3 WHERE orderkey = 11; +``` + +After a subsequent checkpoint completes, run the following statements in the Flink SQL Client: + +```sql +DESCRIBE ods.orders; + +SELECT orderkey, total_price, quantity +FROM ods.orders +ORDER BY orderkey; +``` + +The target schema should contain `quantity BIGINT`, with the following values: + +| orderkey | total_price | quantity | +|----------|-------------|----------| +| 1 | 100.00 | 2 | +| 2 | 200.00 | NULL | +| 11 | 300.00 | 3 | +| 12 | 400.00 | NULL | + +## Synchronize inserts, updates, and deletes + +In the MySQL client, insert an order, update an existing order, and delete another: + +```sql +INSERT INTO orders.orders_2 VALUES (13, 3, 500.00, '2026-01-05', 1); +UPDATE orders.orders_1 +SET total_price = 120.00, quantity = 4 +WHERE orderkey = 1; +DELETE FROM orders.orders_1 WHERE orderkey = 2; +``` + +After the changes are committed by a checkpoint, query Paimon again: + +```sql +SELECT orderkey, custkey, total_price, quantity +FROM ods.orders +ORDER BY orderkey; +``` + +The result should contain the inserted order, the updated price and quantity, and no row for order 2: + +| orderkey | custkey | total_price | quantity | +|----------|---------|-------------|----------| +| 1 | 1 | 120.00 | 4 | +| 11 | 1 | 300.00 | 3 | +| 12 | 3 | 400.00 | NULL | +| 13 | 3 | 500.00 | 1 | + +## Clean up + +Exit the SQL and MySQL clients. Replace `<job-id>` with the ingestion Job ID, cancel the job, and stop the tutorial services: + +```shell +"$FLINK_HOME/bin/flink" cancel <job-id> +"$FLINK_HOME/bin/stop-cluster.sh" +docker compose down +``` + +The warehouse files remain under `/tmp/flink-cdc-paimon-tutorial/warehouse`. +Use an empty warehouse when repeating this example from scratch. Review Comment: This is unnecessary. ########## docs/content/docs/examples-and-best-practices/metadata-and-soft-deletes.md: ########## @@ -0,0 +1,377 @@ +--- +title: "Metadata Columns, Computed Columns, and Soft Deletes" +slug: "/docs/examples-and-best-practices/metadata-and-soft-deletes" +sidebar_position: 6 +hide_title: true +displayed_sidebar: docs +--- +<!-- +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + +http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +KIND, either express or implied. See the License for the +specific language governing permissions and limitations +under the License. +--> + +# Metadata Columns, Computed Columns, and Soft Deletes + +A Flink CDC Transform rule can enrich records as they are ingested. +This example reads MySQL orders, adds source identity and operation metadata, computes a date partition, +and writes the result to a Paimon primary key table. + +When a source row is deleted, the pipeline keeps its last values in Paimon with a deletion marker. +A later insert with the same target primary key makes the row active again. + +## Prepare the Flink environment + +Prepare a Linux or macOS machine with Docker Compose, Java 11, and a Flink standalone cluster. +Choose a Flink distribution and the corresponding dependencies from the following table: + +| Flink distribution | Required dependencies | +|--------------------|-----------------------| +| [Flink 1.20.5](https://archive.apache.org/dist/flink/flink-1.20.5/flink-1.20.5-bin-scala_2.12.tgz) | [Flink CDC @@VERSION@@](https://archive.apache.org/dist/flink/flink-cdc-@@VERSION@@/flink-cdc-@@VERSION@@-1.20-bin.tar.gz)<br />[MySQL Pipeline connector](https://repo.maven.apache.org/maven2/org/apache/flink/flink-cdc-pipeline-connector-mysql/@@VERSION@@-1.20/flink-cdc-pipeline-connector-mysql-@@VERSION@@-1.20.jar)<br />[Paimon Pipeline connector](https://repo.maven.apache.org/maven2/org/apache/flink/flink-cdc-pipeline-connector-paimon/@@VERSION@@-1.20/flink-cdc-pipeline-connector-paimon-@@VERSION@@-1.20.jar)<br />[MySQL JDBC 8.0.27](https://repo.maven.apache.org/maven2/mysql/mysql-connector-java/8.0.27/mysql-connector-java-8.0.27.jar)<br />[Hadoop 2.8.3-10.0](https://repo.maven.apache.org/maven2/org/apache/flink/flink-shaded-hadoop-2-uber/2.8.3-10.0/flink-shaded-hadoop-2-uber-2.8.3-10.0.jar)<br />[Paimon SQL 2.1.0 for Flink 1.20](https://repo.maven.apache.org/maven2/org/apache/pa imon/paimon-flink-1.20/2.1.0/paimon-flink-1.20-2.1.0.jar) | +| [Flink 2.2.1](https://archive.apache.org/dist/flink/flink-2.2.1/flink-2.2.1-bin-scala_2.12.tgz) | [Flink CDC @@VERSION@@](https://archive.apache.org/dist/flink/flink-cdc-@@VERSION@@/flink-cdc-@@VERSION@@-2.2-bin.tar.gz)<br />[MySQL Pipeline connector](https://repo.maven.apache.org/maven2/org/apache/flink/flink-cdc-pipeline-connector-mysql/@@VERSION@@-2.2/flink-cdc-pipeline-connector-mysql-@@VERSION@@-2.2.jar)<br />[Paimon Pipeline connector](https://repo.maven.apache.org/maven2/org/apache/flink/flink-cdc-pipeline-connector-paimon/@@VERSION@@-2.2/flink-cdc-pipeline-connector-paimon-@@VERSION@@-2.2.jar)<br />[MySQL JDBC 8.0.27](https://repo.maven.apache.org/maven2/mysql/mysql-connector-java/8.0.27/mysql-connector-java-8.0.27.jar)<br />[Hadoop 2.8.3-10.0](https://repo.maven.apache.org/maven2/org/apache/flink/flink-shaded-hadoop-2-uber/2.8.3-10.0/flink-shaded-hadoop-2-uber-2.8.3-10.0.jar)<br />[Paimon SQL 2.1.0 for Flink 2.1](https://repo.maven.apache.org/maven2/org/apache/paimon/paim on-flink-2.1/2.1.0/paimon-flink-2.1-2.1.0.jar) | + +The Paimon SQL connector in each row matches the Paimon version bundled with the corresponding Pipeline connector. +Download and extract the selected distributions, then set the following paths to your installation directories: + +```shell +export FLINK_HOME=/path/to/flink +export FLINK_CDC_HOME=/path/to/flink-cdc +``` + +Place the MySQL and Paimon Pipeline connector JARs from the table in `$FLINK_CDC_HOME/lib`. + +Place the MySQL JDBC driver from the table in `$FLINK_HOME/lib` before starting Flink. +Save the Hadoop JAR as `$FLINK_HOME/opt/flink-shaded-hadoop.jar` and the Paimon SQL connector +as `$FLINK_HOME/opt/paimon-sql-connector.jar`. +The commands below add Hadoop with `--jar` when submitting the pipeline and verification queries. +If this Hadoop JAR is already in `$FLINK_HOME/lib`, move it to the `opt` path above before starting the cluster. +Loading it as a job dependency avoids conflicts with the libraries used by Flink's own CLI. + +In `$FLINK_HOME/conf/config.yaml`, add or update the following settings. +Merge them into existing `execution`, `jobmanager`, and `taskmanager` sections if those sections already exist: + +```yaml +execution: + checkpointing: + interval: 3s +jobmanager: + memory: + process: + size: 2048m + jvm-metaspace: + size: 512m +taskmanager: + memory: + process: + size: 2048m + jvm-metaspace: + size: 512m + numberOfTaskSlots: 4 +``` + +Checkpointing makes Paimon commits visible to readers. +Four task slots allow the ingestion pipeline and verification queries to run together. +These memory settings accommodate the verification queries in this walkthrough. +Start the cluster: + +```shell +"$FLINK_HOME/bin/start-cluster.sh" +``` + +The [Flink Web UI](http://localhost:8081/) should show an available TaskManager. + +## Prepare the source data + +Save the following content as `compose.yaml` in a new working directory: + +```yaml +services: + mysql: + image: mysql:8.0 + platform: linux/amd64 + ports: + - "3306:3306" + environment: + MYSQL_ROOT_PASSWORD: tutorial + MYSQL_ROOT_HOST: "%" + command: + - --server-id=1 + - --log-bin=mysql-bin + - --binlog-format=ROW + - --binlog-row-image=FULL + - --default-time-zone=+00:00 + healthcheck: + test: ["CMD", "mysqladmin", "ping", "-h", "127.0.0.1", "-uroot", "-ptutorial"] + interval: 2s + timeout: 5s + retries: 30 +``` + +Start MySQL and open its client: + +```shell +docker compose up -d --wait +docker compose exec mysql mysql -uroot -ptutorial +``` + +Create the source table and two orders: + +```sql +CREATE DATABASE inventory; +USE inventory; + +CREATE TABLE orders ( + id BIGINT NOT NULL, + product VARCHAR(100) NOT NULL, + price DECIMAL(10, 2) NOT NULL, + created_at DATETIME(3) NOT NULL, + PRIMARY KEY (id) +); +INSERT INTO orders VALUES + (101, 'book', 29.90, '2026-01-01 12:00:00'), + (102, 'pen', 5.00, '2026-01-02 09:30:00'); +``` + +Keep this client open for the changes below. Run shell commands in another terminal. + +## Define the Transform rule + +Save the following configuration as `transform.yaml`: + +```yaml +source: + type: mysql + hostname: localhost + port: 3306 + username: root + password: tutorial + tables: inventory.orders + server-id: 5400-5404 + server-time-zone: UTC + metadata.list: op_ts + +transform: + - source-table: inventory.orders + projection: >- + *, + CONCAT_WS('.', __namespace_name__, __schema_name__, __table_name__) AS source_table, + op_ts AS source_ts, + __data_event_type__ AS op_type, + UPPER(product) AS product_upper, + DATE_FORMAT(created_at, 'yyyyMMdd') AS order_day + primary-keys: id, order_day + partition-keys: order_day + converter-after-transform: SOFT_DELETE + +sink: + type: paimon + catalog.properties.metastore: filesystem + catalog.properties.warehouse: file:///tmp/flink-cdc-transform/warehouse + +pipeline: + name: Enrich MySQL orders with soft deletes + parallelism: 2 + local-time-zone: UTC + schema.change.behavior: evolve +``` + +The folded YAML scalar after `projection: >-` allows the expression to span multiple lines. +The `*` projects all source columns, followed by the additional columns: + +| Column | Expression | Meaning | +|--------|------------|---------| +| `source_table` | `CONCAT_WS('.', __namespace_name__, __schema_name__, __table_name__)` | Builds the source identifier while skipping null hierarchy levels. This example produces `inventory.orders`. | +| `source_ts` | `op_ts` | Source operation time in epoch milliseconds. Requires `metadata.list: op_ts`; snapshot records have a value of `0`. | +| `op_type` | `__data_event_type__` | The original change type, retained when a delete is converted to an insert. | +| `product_upper` | `UPPER(product)` | An uppercase product name. | +| `order_day` | `DATE_FORMAT(created_at, 'yyyyMMdd')` | The order date used to partition the target table. | + +Paimon partition columns must be included in the primary key, so the target key is `id, order_day`. +This example keeps `created_at` unchanged, making the derived partition key stable. +The warehouse uses a local path because all Flink processes run on one machine; +use shared storage when deploying to multiple machines. + +`SOFT_DELETE` runs after the projection and converts a delete event into an insert containing the projected old row. +The stored `op_type` still identifies the source delete. +The main operation markers are: + +| Marker | Meaning | +|--------|---------| +| `+I` | Snapshot row or insert. | +| `-U` | Before image of an update. | +| `+U` | After image of an update. | +| `-D` | Delete; the row is retained by this pipeline's soft-delete converter. | + +The Paimon primary key table stores the latest state for each target key. +Readers can select active rows by excluding the `-D` marker. +For a pipeline that publishes individual change events, see [Real-time MySQL Change Distribution to Kafka](mysql-to-kafka.md). + +Submit the pipeline and record its Job ID: + +```shell +"$FLINK_CDC_HOME/bin/flink-cdc.sh" transform.yaml \ + --jar "$FLINK_HOME/opt/flink-shaded-hadoop.jar" +``` + +Wait for the initial snapshot and a completed checkpoint. + +## Verify metadata and computed columns + +Open the SQL Client: + +```shell +"$FLINK_HOME/bin/sql-client.sh" \ + --jar "$FLINK_HOME/opt/paimon-sql-connector.jar" \ + --jar "$FLINK_HOME/opt/flink-shaded-hadoop.jar" +``` + +Create a catalog pointing to the same warehouse and inspect the target: + +```sql +SET 'execution.runtime-mode' = 'batch'; +SET 'sql-client.execution.result-mode' = 'tableau'; +SET 'parallelism.default' = '1'; + +CREATE CATALOG lake WITH ( + 'type' = 'paimon', + 'metastore' = 'filesystem', + 'warehouse' = 'file:///tmp/flink-cdc-transform/warehouse', + 'cache-enabled' = 'false' +); +USE CATALOG lake; + +SHOW CREATE TABLE inventory.orders; + +SELECT id, product_upper, price, order_day, op_type +FROM inventory.orders +ORDER BY id; + +SELECT id, source_table, source_ts +FROM inventory.orders +ORDER BY id; +``` + +The table definition should contain the added columns, the `id, order_day` primary key, and the `order_day` partition. + +The first query should return: + +| id | product_upper | price | order_day | op_type | +|----|---------------|-------|-----------|---------| +| 101 | BOOK | 29.90 | 20260101 | +I | +| 102 | PEN | 5.00 | 20260102 | +I | + +The metadata query should return: + +| id | source_table | source_ts | +|----|--------------|-----------| +| 101 | inventory.orders | 0 | +| 102 | inventory.orders | 0 | + +`source_ts` records database operation time; `created_at` is the business timestamp used for partitioning. +The zero value for snapshot rows does not represent their business creation time. + +## Verify updates and soft deletes + +Run these statements in MySQL: + +```sql +UPDATE inventory.orders +SET product = 'notebook', price = 39.90 +WHERE id = 101; +DELETE FROM inventory.orders WHERE id = 102; +INSERT INTO inventory.orders VALUES + (103, 'pencil', 3.50, '2026-01-03 12:00:00'); +``` + +After the changes and a subsequent checkpoint complete, query all rows and then the active rows: + +```sql +SELECT id, product_upper, price, order_day, op_type, + source_ts > 0 AS has_binlog_time +FROM inventory.orders +ORDER BY id; + +SELECT id, product_upper, price +FROM inventory.orders +WHERE op_type <> '-D' +ORDER BY id; +``` + +The full result should retain the deleted order: + +| id | product_upper | price | order_day | op_type | has_binlog_time | +|----|---------------|-------|-----------|---------|-----------------| +| 101 | NOTEBOOK | 39.90 | 20260101 | +U | TRUE | +| 102 | PEN | 5.00 | 20260102 | -D | TRUE | +| 103 | PENCIL | 3.50 | 20260103 | +I | TRUE | + +The active-row query should contain only: + +| id | product_upper | price | +|----|---------------|-------| +| 101 | NOTEBOOK | 39.90 | +| 103 | PENCIL | 3.50 | + +Order 102 has been physically deleted from MySQL. +In Paimon, its last values remain under the same target key with `op_type = '-D'`. +The nonzero operation timestamps confirm that these rows reflect binlog changes. + +## Reinsert a deleted key + +Insert order 102 again, keeping its original partition date: + +```sql +INSERT INTO inventory.orders VALUES + (102, 'blue pen', 6.00, '2026-01-02 09:30:00'); +``` + +After a subsequent checkpoint completes, query the target: + +```sql +SELECT id, product_upper, price, order_day, op_type, + source_ts > 0 AS has_binlog_time +FROM inventory.orders +ORDER BY id; + +SELECT id, product_upper, price +FROM inventory.orders +WHERE op_type <> '-D' +ORDER BY id; +``` + +The new insert should replace the deletion marker for the same `id, order_day` key: + +| id | product_upper | price | order_day | op_type | has_binlog_time | +|----|---------------|-------|-----------|---------|-----------------| +| 101 | NOTEBOOK | 39.90 | 20260101 | +U | TRUE | +| 102 | BLUE PEN | 6.00 | 20260102 | +I | TRUE | +| 103 | PENCIL | 3.50 | 20260103 | +I | TRUE | + +The active-row query now includes all three orders. +For more expressions and metadata fields, see [Transform](../core-concept/transform.md). + +## Clean up + +Replace `<job-id>` with the ingestion Job ID, cancel the job, and stop the tutorial services: + +```shell +"$FLINK_HOME/bin/flink" cancel <job-id> +"$FLINK_HOME/bin/stop-cluster.sh" +docker compose down +``` + +The warehouse remains under `/tmp/flink-cdc-transform/warehouse`. +Use an empty warehouse when starting this example again from scratch. Review Comment: ditto. ########## docs/i18n/zh/docusaurus-plugin-content-docs/current/docs/examples-and-best-practices/mysql-to-paimon.md: ########## @@ -0,0 +1,301 @@ +--- +title: "MySQL 数据实时入湖 Paimon" +slug: "/docs/examples-and-best-practices/mysql-to-paimon" +sidebar_position: 2 +hide_title: true +displayed_sidebar: docs +--- +<!-- +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + +http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +KIND, either express or implied. See the License for the +specific language governing permissions and limitations +under the License. +--> + +# MySQL 数据实时入湖 Paimon + +本示例通过 Flink CDC YAML 数据摄入作业，将 MySQL 中的两张订单分表合并写入一张 Apache Paimon 表， +演示全量快照、表结构变更，以及后续插入、更新和删除的同步。文中的 SQL 查询用于验证摄入结果。 + +两张源表具有相同的表结构，且订单主键在两表之间不重复。 +如果需要合并的分表存在重复主键，应将来源表标识纳入联合主键； +否则，相同主键的记录可能在 Paimon 中相互覆盖。 + +## 准备 Flink 环境 + +准备一台安装了 Docker Compose 和 Java 11 的 Linux 或 macOS 计算机，以及一个 Flink Standalone 集群。 +根据下表选择 Flink 发行包及其对应的依赖： + +| Flink 发行包 | 所需依赖 | +|--------------|----------| +| [Flink 1.20.5](https://archive.apache.org/dist/flink/flink-1.20.5/flink-1.20.5-bin-scala_2.12.tgz) | [Flink CDC @@VERSION@@](https://archive.apache.org/dist/flink/flink-cdc-@@VERSION@@/flink-cdc-@@VERSION@@-1.20-bin.tar.gz)<br />[MySQL Pipeline 连接器](https://repo.maven.apache.org/maven2/org/apache/flink/flink-cdc-pipeline-connector-mysql/@@VERSION@@-1.20/flink-cdc-pipeline-connector-mysql-@@VERSION@@-1.20.jar)<br />[Paimon Pipeline 连接器](https://repo.maven.apache.org/maven2/org/apache/flink/flink-cdc-pipeline-connector-paimon/@@VERSION@@-1.20/flink-cdc-pipeline-connector-paimon-@@VERSION@@-1.20.jar)<br />[MySQL JDBC 8.0.27](https://repo.maven.apache.org/maven2/mysql/mysql-connector-java/8.0.27/mysql-connector-java-8.0.27.jar)<br />[Hadoop 2.8.3-10.0](https://repo.maven.apache.org/maven2/org/apache/flink/flink-shaded-hadoop-2-uber/2.8.3-10.0/flink-shaded-hadoop-2-uber-2.8.3-10.0.jar)<br />[面向 Flink 1.20 的 Paimon SQL 2.1.0](https://repo.maven.apache.org/maven2/org/ap ache/paimon/paimon-flink-1.20/2.1.0/paimon-flink-1.20-2.1.0.jar) | +| [Flink 2.2.1](https://archive.apache.org/dist/flink/flink-2.2.1/flink-2.2.1-bin-scala_2.12.tgz) | [Flink CDC @@VERSION@@](https://archive.apache.org/dist/flink/flink-cdc-@@VERSION@@/flink-cdc-@@VERSION@@-2.2-bin.tar.gz)<br />[MySQL Pipeline 连接器](https://repo.maven.apache.org/maven2/org/apache/flink/flink-cdc-pipeline-connector-mysql/@@VERSION@@-2.2/flink-cdc-pipeline-connector-mysql-@@VERSION@@-2.2.jar)<br />[Paimon Pipeline 连接器](https://repo.maven.apache.org/maven2/org/apache/flink/flink-cdc-pipeline-connector-paimon/@@VERSION@@-2.2/flink-cdc-pipeline-connector-paimon-@@VERSION@@-2.2.jar)<br />[MySQL JDBC 8.0.27](https://repo.maven.apache.org/maven2/mysql/mysql-connector-java/8.0.27/mysql-connector-java-8.0.27.jar)<br />[Hadoop 2.8.3-10.0](https://repo.maven.apache.org/maven2/org/apache/flink/flink-shaded-hadoop-2-uber/2.8.3-10.0/flink-shaded-hadoop-2-uber-2.8.3-10.0.jar)<br />[面向 Flink 2.1 的 Paimon SQL 2.1.0](https://repo.maven.apache.org/maven2/org/apache/paim on/paimon-flink-2.1/2.1.0/paimon-flink-2.1-2.1.0.jar) | + +表中 Paimon SQL 连接器的版本与对应 Pipeline 连接器内置的 Paimon 版本一致。 +下载并解压选定的发行包，将以下路径替换为实际安装目录： + +```shell +export FLINK_HOME=/path/to/flink +export FLINK_CDC_HOME=/path/to/flink-cdc +``` + +将表中的 MySQL 和 Paimon Pipeline 连接器 JAR 放入 `$FLINK_CDC_HOME/lib`。 + +启动 Flink 前，将表中的 MySQL JDBC 驱动放入 `$FLINK_HOME/lib`。 +将 Hadoop JAR 保存为 `$FLINK_HOME/opt/flink-shaded-hadoop.jar`， +将 Paimon SQL 连接器保存为 `$FLINK_HOME/opt/paimon-sql-connector.jar`。 +下文在提交摄入作业和验证查询时，通过 `--jar` 附加 Hadoop 依赖。 +如果 `$FLINK_HOME/lib` 中已有该 Hadoop JAR，请在启动集群前将其移至上述 `opt` 路径。 +按作业加载该依赖，可以避免它与 Flink 自身命令行所用的依赖发生冲突。 + +在 `$FLINK_HOME/conf/config.yaml` 中添加或更新以下配置。 +如果文件中已有 `execution` 和 `taskmanager` 节点，请将配置合并到对应节点中： + +```yaml +execution: + checkpointing: + interval: 3s +taskmanager: + numberOfTaskSlots: 4 +``` + +Checkpoint 用于提交 Paimon 数据，使其对查询可见。 +配置四个 Task Slot，以便摄入作业和验证查询同时运行。 +启动集群： + +```shell +"$FLINK_HOME/bin/start-cluster.sh" +``` + +在 [Flink Web UI](http://localhost:8081/) 中确认已有可用的 TaskManager。 + +## 准备 MySQL 数据表 + +新建一个工作目录，将以下内容保存为 `compose.yaml`： + +```yaml +services: + mysql: + image: mysql:8.0 + platform: linux/amd64 + ports: + - "3306:3306" + environment: + MYSQL_ROOT_PASSWORD: tutorial + MYSQL_ROOT_HOST: "%" + command: + - --server-id=1 + - --log-bin=mysql-bin + - --binlog-format=ROW + - --binlog-row-image=FULL + - --default-time-zone=+00:00 + healthcheck: + test: ["CMD", "mysqladmin", "ping", "-h", "127.0.0.1", "-uroot", "-ptutorial"] + interval: 2s + timeout: 5s + retries: 30 +``` + +启动 MySQL，等待健康检查通过，然后进入 MySQL 客户端： + +```shell +docker compose up -d --wait +docker compose exec mysql mysql -uroot -ptutorial +``` + +在 MySQL 客户端中创建两张源表并写入测试数据： + +```sql +CREATE DATABASE orders; +USE orders; + +CREATE TABLE orders_1 ( + orderkey BIGINT NOT NULL, + custkey BIGINT, + total_price DECIMAL(10, 2), + order_date DATE, + PRIMARY KEY (orderkey) +); +CREATE TABLE orders_2 LIKE orders_1; + +INSERT INTO orders_1 VALUES + (1, 1, 100.00, '2026-01-01'), + (2, 2, 200.00, '2026-01-02'); +INSERT INTO orders_2 VALUES + (11, 1, 300.00, '2026-01-03'), + (12, 3, 400.00, '2026-01-04'); +``` + +保留此客户端，供后续执行表结构和数据变更。其余 Shell 命令请在另一个终端中执行。 + +## 配置并提交数据摄入作业 + +将以下配置保存为 `mysql-to-paimon.yaml`： + +```yaml +source: + type: mysql + hostname: localhost + port: 3306 + username: root + password: tutorial + tables: orders.orders_[0-9]+ + server-id: 5400-5404 + server-time-zone: UTC + +sink: + type: paimon + catalog.properties.metastore: filesystem + catalog.properties.warehouse: file:///tmp/flink-cdc-paimon-tutorial/warehouse + +route: + - source-table: orders.orders_[0-9]+ + sink-table: ods.orders + +pipeline: + name: MySQL orders to Paimon + parallelism: 2 + schema.change.behavior: evolve +``` + +`tables` 表达式选中两张订单表，路由规则将其合并到 `ods.orders`。 +Flink CDC 会自动创建目标数据库和表。 + +本示例的 Flink 进程运行在同一台计算机上，因此使用 `file:///` 形式的本地仓库路径。 +部署到分布式集群时，应使用客户端、JobManager 和所有 TaskManager 都能访问的共享存储。 + +提交作业： + +```shell +"$FLINK_CDC_HOME/bin/flink-cdc.sh" mysql-to-paimon.yaml \ + --jar "$FLINK_HOME/opt/flink-shaded-hadoop.jar" +``` + +记下返回的 Job ID。在 Flink Web UI 中确认作业正常运行，且至少完成一次 Checkpoint。 + +## 验证全量快照 + +启动 Flink SQL Client： + +```shell +"$FLINK_HOME/bin/sql-client.sh" \ + --jar "$FLINK_HOME/opt/paimon-sql-connector.jar" \ + --jar "$FLINK_HOME/opt/flink-shaded-hadoop.jar" +``` + +创建指向同一仓库路径的 Paimon Catalog，并查询合并后的目标表： + +```sql +SET 'execution.runtime-mode' = 'batch'; +SET 'sql-client.execution.result-mode' = 'tableau'; +SET 'parallelism.default' = '1'; + +CREATE CATALOG lake WITH ( + 'type' = 'paimon', + 'metastore' = 'filesystem', + 'warehouse' = 'file:///tmp/flink-cdc-paimon-tutorial/warehouse', + 'cache-enabled' = 'false' +); +USE CATALOG lake; + +SELECT orderkey, custkey, total_price, order_date +FROM ods.orders +ORDER BY orderkey; +``` + +结果应包含以下四条记录： + +| orderkey | custkey | total_price | order_date | +|----------|---------|-------------|------------| +| 1 | 1 | 100.00 | 2026-01-01 | +| 2 | 2 | 200.00 | 2026-01-02 | +| 11 | 1 | 300.00 | 2026-01-03 | +| 12 | 3 | 400.00 | 2026-01-04 | + +如果目标表尚不可用，请等待全量快照和一次成功的 Checkpoint，再执行查询。 +批查询读取已提交的快照；后续变更需要重新执行查询才能看到。 + +## 同步表结构变更 + +在 MySQL 客户端中，为两张源表增加同一字段，并更新其中两条记录： + +```sql +ALTER TABLE orders.orders_1 ADD COLUMN quantity BIGINT; +ALTER TABLE orders.orders_2 ADD COLUMN quantity BIGINT; +UPDATE orders.orders_1 SET quantity = 2 WHERE orderkey = 1; +UPDATE orders.orders_2 SET quantity = 3 WHERE orderkey = 11; +``` + +等待后续 Checkpoint 完成，然后在 Flink SQL Client 中执行： + +```sql +DESCRIBE ods.orders; + +SELECT orderkey, total_price, quantity +FROM ods.orders +ORDER BY orderkey; +``` + +目标表结构应包含 `quantity BIGINT`，对应的数据如下： + +| orderkey | total_price | quantity | +|----------|-------------|----------| +| 1 | 100.00 | 2 | +| 2 | 200.00 | NULL | +| 11 | 300.00 | 3 | +| 12 | 400.00 | NULL | + +## 同步插入、更新与删除 + +在 MySQL 客户端中插入新订单、更新已有订单，并删除另一条订单： + +```sql +INSERT INTO orders.orders_2 VALUES (13, 3, 500.00, '2026-01-05', 1); +UPDATE orders.orders_1 +SET total_price = 120.00, quantity = 4 +WHERE orderkey = 1; +DELETE FROM orders.orders_1 WHERE orderkey = 2; +``` + +等待这些变更通过 Checkpoint 提交后，再次查询 Paimon： + +```sql +SELECT orderkey, custkey, total_price, quantity +FROM ods.orders +ORDER BY orderkey; +``` + +结果应包含新增订单、更新后的价格和数量，且不再包含订单 2： + +| orderkey | custkey | total_price | quantity | +|----------|---------|-------------|----------| +| 1 | 1 | 120.00 | 4 | +| 11 | 1 | 300.00 | 3 | +| 12 | 3 | 400.00 | NULL | +| 13 | 3 | 500.00 | 1 | + +## 清理环境 + +退出 SQL Client 和 MySQL 客户端，将 `<job-id>` 替换为摄入作业的 Job ID， +取消作业并停止本示例的服务： + +```shell +"$FLINK_HOME/bin/flink" cancel <job-id> +"$FLINK_HOME/bin/stop-cluster.sh" +docker compose down +``` + +仓库文件保留在 `/tmp/flink-cdc-paimon-tutorial/warehouse`。 +如果需要从头重做本示例，请使用空仓库。 Review Comment: ditto. ########## docs/content/docs/examples-and-best-practices/ai-text-labeling.md: ########## @@ -0,0 +1,321 @@ +--- +title: "AI Function Configuration and Real-time Text Labeling" +slug: "/docs/examples-and-best-practices/ai-text-labeling" +sidebar_position: 7 +hide_title: true +displayed_sidebar: docs +--- +<!-- +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + +http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +KIND, either express or implied. See the License for the +specific language governing permissions and limitations +under the License. +--> + +# AI Function Configuration and Real-time Text Labeling + +This article shows how to use AI Functions in a Flink CDC YAML ingestion pipeline +to classify MySQL customer feedback, flag escalation risk, generate summaries, +and mask contact details before writing the results to Paimon. + +## Prepare the Flink environment + +Prepare a Linux or macOS machine with Docker Compose, Java 11, and a Flink standalone cluster. +Choose a Flink distribution and the corresponding dependencies from the following table: + +| Flink distribution | Required dependencies | +|--------------------|-----------------------| +| [Flink 1.20.5](https://archive.apache.org/dist/flink/flink-1.20.5/flink-1.20.5-bin-scala_2.12.tgz) | [Flink CDC @@VERSION@@](https://archive.apache.org/dist/flink/flink-cdc-@@VERSION@@/flink-cdc-@@VERSION@@-1.20-bin.tar.gz)<br />[MySQL Pipeline connector](https://repo.maven.apache.org/maven2/org/apache/flink/flink-cdc-pipeline-connector-mysql/@@VERSION@@-1.20/flink-cdc-pipeline-connector-mysql-@@VERSION@@-1.20.jar)<br />[Paimon Pipeline connector](https://repo.maven.apache.org/maven2/org/apache/flink/flink-cdc-pipeline-connector-paimon/@@VERSION@@-1.20/flink-cdc-pipeline-connector-paimon-@@VERSION@@-1.20.jar)<br />[AI model client](https://repo.maven.apache.org/maven2/org/apache/flink/flink-cdc-pipeline-model-openai-compatible/@@VERSION@@-1.20/flink-cdc-pipeline-model-openai-compatible-@@VERSION@@-1.20.jar)<br />[MySQL JDBC 8.0.27](https://repo.maven.apache.org/maven2/mysql/mysql-connector-java/8.0.27/mysql-connector-java-8.0.27.jar)<br />[Hadoop 2.8.3-10.0](https://repo.maven.apa che.org/maven2/org/apache/flink/flink-shaded-hadoop-2-uber/2.8.3-10.0/flink-shaded-hadoop-2-uber-2.8.3-10.0.jar) | +| [Flink 2.2.1](https://archive.apache.org/dist/flink/flink-2.2.1/flink-2.2.1-bin-scala_2.12.tgz) | [Flink CDC @@VERSION@@](https://archive.apache.org/dist/flink/flink-cdc-@@VERSION@@/flink-cdc-@@VERSION@@-2.2-bin.tar.gz)<br />[MySQL Pipeline connector](https://repo.maven.apache.org/maven2/org/apache/flink/flink-cdc-pipeline-connector-mysql/@@VERSION@@-2.2/flink-cdc-pipeline-connector-mysql-@@VERSION@@-2.2.jar)<br />[Paimon Pipeline connector](https://repo.maven.apache.org/maven2/org/apache/flink/flink-cdc-pipeline-connector-paimon/@@VERSION@@-2.2/flink-cdc-pipeline-connector-paimon-@@VERSION@@-2.2.jar)<br />[AI model client](https://repo.maven.apache.org/maven2/org/apache/flink/flink-cdc-pipeline-model-openai-compatible/@@VERSION@@-2.2/flink-cdc-pipeline-model-openai-compatible-@@VERSION@@-2.2.jar)<br />[MySQL JDBC 8.0.27](https://repo.maven.apache.org/maven2/mysql/mysql-connector-java/8.0.27/mysql-connector-java-8.0.27.jar)<br />[Hadoop 2.8.3-10.0](https://repo.maven.apache.org/ma ven2/org/apache/flink/flink-shaded-hadoop-2-uber/2.8.3-10.0/flink-shaded-hadoop-2-uber-2.8.3-10.0.jar)<br />[Paimon SQL 2.1.0 for Flink 2.1](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-flink-2.1/2.1.0/paimon-flink-2.1-2.1.0.jar) | + +The Paimon SQL connector for the query environment matches the Paimon version bundled with the Pipeline connector. +Download and extract the selected distributions, then set the following paths to your installation directories: + +```shell +export FLINK_HOME=/path/to/flink +export FLINK_CDC_HOME=/path/to/flink-cdc +``` + +Place the MySQL and Paimon Pipeline connectors and AI model client JARs from the table in `$FLINK_CDC_HOME/lib`. + +Place the MySQL JDBC driver from the table in `$FLINK_HOME/lib` before starting Flink. +Save the Hadoop JAR as `$FLINK_HOME/opt/flink-shaded-hadoop.jar`. +In the query environment, save the Paimon SQL connector as `$FLINK_HOME/opt/paimon-sql-connector.jar`. +The commands below add Hadoop with `--jar` when submitting the pipeline and verification queries. +If this Hadoop JAR is already in `$FLINK_HOME/lib`, move it to the `opt` path above before starting the cluster. +Loading it as a job dependency avoids conflicts with the libraries used by Flink's own CLI. + +In `$FLINK_HOME/conf/config.yaml`, add or update the following settings. +Merge them into existing `execution`, `jobmanager`, and `taskmanager` sections if those sections already exist: + +```yaml +execution: + checkpointing: + interval: 3s +jobmanager: + memory: + process: + size: 2048m + jvm-metaspace: + size: 512m +taskmanager: + memory: + process: + size: 2048m + jvm-metaspace: + size: 512m + numberOfTaskSlots: 4 +``` + +Checkpointing makes Paimon commits visible to readers. +Four task slots allow the ingestion pipeline and verification queries to run together. +These memory settings accommodate the verification queries in this walkthrough. +Start the cluster: + +```shell +"$FLINK_HOME/bin/start-cluster.sh" +``` + +The [Flink Web UI](http://localhost:8081/) should show an available TaskManager. + +## Prepare the model service + +Prepare a model endpoint that supports the OpenAI-compatible `chat/completions` protocol and JSON-object responses. +The Flink client and TaskManagers must be able to reach it. + +In the pipeline below, replace `endpoint`, `model`, and `api-key` with the actual service settings. +The endpoint is the service's base URL, such as `https://api.example.com/v1`, rather than the complete completion path. +Use the model name offered by that service. + +Supply the credential through your deployment's configuration mechanism. +The value in this example is a placeholder. + +## Prepare customer feedback + +Save the following content as `compose.yaml` in a new working directory: + +```yaml +services: + mysql: + image: mysql:8.0 + platform: linux/amd64 + ports: + - "3306:3306" + environment: + MYSQL_ROOT_PASSWORD: tutorial + MYSQL_ROOT_HOST: "%" + command: + - --server-id=1 + - --log-bin=mysql-bin + - --binlog-format=ROW + - --binlog-row-image=FULL + - --default-time-zone=+00:00 + healthcheck: + test: ["CMD", "mysqladmin", "ping", "-h", "127.0.0.1", "-uroot", "-ptutorial"] + interval: 2s + timeout: 5s + retries: 30 +``` + +Start MySQL and open its client: + +```shell +docker compose up -d --wait +docker compose exec mysql mysql -uroot -ptutorial +``` + +Create the source table and two feedback records: + +```sql +CREATE DATABASE inventory; +USE inventory; + +CREATE TABLE feedback ( + id BIGINT NOT NULL, + content TEXT NOT NULL, + PRIMARY KEY (id) +); +INSERT INTO feedback VALUES + (101, 'My blender broke after one day. I will post a public complaint. Email me at [email protected].'), + (102, 'My order arrived three days late. Email me at [email protected].'); +``` + +Keep the client open for subsequent changes. Run shell commands in another terminal. + +## Configure the model and AI functions + +Save the following configuration as `ai-labeling.yaml` and fill in the model service settings: + +```yaml +source: + type: mysql + hostname: localhost + port: 3306 + username: root + password: tutorial + tables: inventory.feedback + server-id: 5400-5404 + server-time-zone: UTC + +transform: + - source-table: inventory.feedback + projection: >- + id, + AI_CLASSIFY('text_model', content, 'quality,delivery,service,other')['category'] AS category, + AI_COMPLETE('text_model', content, 'Return high if the feedback explicitly threatens escalation or public exposure; otherwise return low. Put only high or low in result.')['result'] AS risk, + AI_SUMMARIZE('text_model', content, 80)['summary'] AS summary, + AI_MASK('text_model', content, 'email,phone')['masked_text'] AS masked_content + +sink: + type: paimon + catalog.properties.metastore: filesystem + catalog.properties.warehouse: file:///tmp/flink-cdc-ai-labeling/warehouse + +pipeline: + name: AI text labeling during ingestion + parallelism: 2 + model: + - name: text_model + type: openai-compatible + options: + model: <model-name> + endpoint: https://api.example.com/v1 + api-key: <api-key> + system-prompt: >- + Use English in generated text. Follow the task-specific JSON shape. + Never include email addresses or phone numbers in summaries. + In masked_text, replace each email address or phone number with [REDACTED]. + temperature: 0 + max-tokens: 256 + response-format: json_object + error-handling-strategy: retry + retry-num: 3 + retry-fallback-strategy: failover + retry-backoff-base-interval: 1 s +``` + +The `pipeline.model` block declares the model client, and the Transform expressions refer to its `name`: + +| Setting | Purpose | +|---------|---------| +| `name: text_model` | The local alias passed as the first argument to each AI function. | +| `type: openai-compatible` | Selects the model client implementation from the downloaded JAR. | +| `options.model` | The actual model name sent to the service. | +| `options.endpoint`, `options.api-key` | The base URL and authentication credential. | +| `options.system-prompt` | Shared instructions added before the function-specific prompt. | +| `response-format: json_object` | Requests a JSON-object response from the service. | +| `retry-num: 3`, `retry-fallback-strategy: failover` | Makes at most three attempts for a failed model request, then fails the task. | + +The four functions cover the labeling scenarios: + +| Function | Task | JSON field used | +|----------|------|-----------------| +| `AI_CLASSIFY` | Choose one of the comma-separated topic labels. | `category` | +| `AI_COMPLETE` | Apply the custom high/low escalation-risk prompt. | `result` | +| `AI_SUMMARIZE` | Produce a summary of at most 80 characters. | `summary` | +| `AI_MASK` | Replace email addresses and phone numbers. | `masked_text` | + +Text AI functions return `VARIANT` values. The `['field']` expression selects the required field, which is written directly to a Paimon `VARIANT` column. + +Only the primary key and selected AI outputs are projected. +The original feedback text and the masking function's detected-entity details are not written to the target table. + +Submit the pipeline and record its Job ID: + +```shell +"$FLINK_CDC_HOME/bin/flink-cdc.sh" ai-labeling.yaml \ + --jar "$FLINK_HOME/opt/flink-shaded-hadoop.jar" +``` + +Wait for the initial model calls and a successful checkpoint. + +## Verify the initial labels + +The queries below use the Flink 2.2 SQL Client. A Flink 1.20 ingestion job can write these columns, +but querying them requires a reader with Paimon `VARIANT` support. + +Open the SQL Client: + +```shell +"$FLINK_HOME/bin/sql-client.sh" \ + --jar "$FLINK_HOME/opt/paimon-sql-connector.jar" \ + --jar "$FLINK_HOME/opt/flink-shaded-hadoop.jar" +``` + +Create a catalog pointing to the same warehouse and query the results: + +```sql +SET 'execution.runtime-mode' = 'batch'; +SET 'sql-client.execution.result-mode' = 'tableau'; +SET 'parallelism.default' = '1'; + +CREATE CATALOG lake WITH ( + 'type' = 'paimon', + 'metastore' = 'filesystem', + 'warehouse' = 'file:///tmp/flink-cdc-ai-labeling/warehouse', + 'cache-enabled' = 'false' +); +USE CATALOG lake; + +SELECT id, + category, + risk, + summary, + masked_content +FROM inventory.feedback +ORDER BY id; +``` + +Check whether the output matches the expected results. + +## Verify incremental labeling + +In the MySQL client, replace the first complaint with resolved feedback, delete the second record, and add another delivery complaint: + +```sql +UPDATE inventory.feedback +SET content = 'The replacement blender works well. Thank you. Email me at [email protected].' +WHERE id = 101; +DELETE FROM inventory.feedback WHERE id = 102; +INSERT INTO inventory.feedback VALUES + (103, 'The courier lost my parcel. Email me at [email protected].'); +``` + +After the changes have been processed and a subsequent checkpoint completes, query Paimon again: + +```sql +SELECT id, + category, + risk, + summary, + masked_content +FROM inventory.feedback +ORDER BY id; +``` + +Check whether the output matches the expected results. + +## Clean up + +Replace `<job-id>` with the ingestion Job ID, cancel the job, and stop the tutorial services: + +```shell +"$FLINK_HOME/bin/flink" cancel <job-id> +"$FLINK_HOME/bin/stop-cluster.sh" +docker compose down +``` + +The warehouse remains under `/tmp/flink-cdc-ai-labeling/warehouse`. +Use an empty warehouse when starting this example again from scratch. Review Comment: ditto. ########## docs/content/docs/examples-and-best-practices/table-selection-and-new-tables.md: ########## @@ -0,0 +1,408 @@ +--- +title: "Table Selection and Adding Tables" +slug: "/docs/examples-and-best-practices/table-selection-and-new-tables" +sidebar_position: 5 +hide_title: true +displayed_sidebar: docs +--- +<!-- +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + +http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +KIND, either express or implied. See the License for the +specific language governing permissions and limitations +under the License. +--> + +# Table Selection and Adding Tables + +A MySQL ingestion pipeline can select tables with regular expressions and exclude temporary tables. +Adding a newly created empty table and adding an existing table with historical data require different source options: + +| Scenario | MySQL source option | How it takes effect | Data captured | +|----------|---------------------|---------------------|---------------| +| A new empty table is created while the job is reading binlog | `scan.binlog.newly-added-table.enabled: true` | The running job captures the table's creation and subsequent changes. | Changes recorded in binlog; no historical snapshot is taken for the new table. | +| An existing table with historical rows is added to the captured set | `scan.newly-added-table.enabled: true` | Update the table selection and restore the job from a savepoint. | A snapshot of the newly included table, followed by its changes. | + +These options cannot both be enabled. +This example first captures a new table without restarting, then restores the same pipeline to include a historical table. +Temporary tables remain excluded throughout. + +## Prepare the Flink environment + +Prepare a Linux or macOS machine with Docker Compose, Java 11, and a Flink standalone cluster. +Choose a Flink distribution and the corresponding dependencies from the following table: + +| Flink distribution | Required dependencies | +|--------------------|-----------------------| +| [Flink 1.20.5](https://archive.apache.org/dist/flink/flink-1.20.5/flink-1.20.5-bin-scala_2.12.tgz) | [Flink CDC @@VERSION@@](https://archive.apache.org/dist/flink/flink-cdc-@@VERSION@@/flink-cdc-@@VERSION@@-1.20-bin.tar.gz)<br />[MySQL Pipeline connector](https://repo.maven.apache.org/maven2/org/apache/flink/flink-cdc-pipeline-connector-mysql/@@VERSION@@-1.20/flink-cdc-pipeline-connector-mysql-@@VERSION@@-1.20.jar)<br />[Paimon Pipeline connector](https://repo.maven.apache.org/maven2/org/apache/flink/flink-cdc-pipeline-connector-paimon/@@VERSION@@-1.20/flink-cdc-pipeline-connector-paimon-@@VERSION@@-1.20.jar)<br />[MySQL JDBC 8.0.27](https://repo.maven.apache.org/maven2/mysql/mysql-connector-java/8.0.27/mysql-connector-java-8.0.27.jar)<br />[Hadoop 2.8.3-10.0](https://repo.maven.apache.org/maven2/org/apache/flink/flink-shaded-hadoop-2-uber/2.8.3-10.0/flink-shaded-hadoop-2-uber-2.8.3-10.0.jar)<br />[Paimon SQL 2.1.0 for Flink 1.20](https://repo.maven.apache.org/maven2/org/apache/pa imon/paimon-flink-1.20/2.1.0/paimon-flink-1.20-2.1.0.jar) | +| [Flink 2.2.1](https://archive.apache.org/dist/flink/flink-2.2.1/flink-2.2.1-bin-scala_2.12.tgz) | [Flink CDC @@VERSION@@](https://archive.apache.org/dist/flink/flink-cdc-@@VERSION@@/flink-cdc-@@VERSION@@-2.2-bin.tar.gz)<br />[MySQL Pipeline connector](https://repo.maven.apache.org/maven2/org/apache/flink/flink-cdc-pipeline-connector-mysql/@@VERSION@@-2.2/flink-cdc-pipeline-connector-mysql-@@VERSION@@-2.2.jar)<br />[Paimon Pipeline connector](https://repo.maven.apache.org/maven2/org/apache/flink/flink-cdc-pipeline-connector-paimon/@@VERSION@@-2.2/flink-cdc-pipeline-connector-paimon-@@VERSION@@-2.2.jar)<br />[MySQL JDBC 8.0.27](https://repo.maven.apache.org/maven2/mysql/mysql-connector-java/8.0.27/mysql-connector-java-8.0.27.jar)<br />[Hadoop 2.8.3-10.0](https://repo.maven.apache.org/maven2/org/apache/flink/flink-shaded-hadoop-2-uber/2.8.3-10.0/flink-shaded-hadoop-2-uber-2.8.3-10.0.jar)<br />[Paimon SQL 2.1.0 for Flink 2.1](https://repo.maven.apache.org/maven2/org/apache/paimon/paim on-flink-2.1/2.1.0/paimon-flink-2.1-2.1.0.jar) | + +The Paimon SQL connector in each row matches the Paimon version bundled with the corresponding Pipeline connector. +Download and extract the selected distributions, then set the following paths to your installation directories: + +```shell +export FLINK_HOME=/path/to/flink +export FLINK_CDC_HOME=/path/to/flink-cdc +``` + +Place the MySQL and Paimon Pipeline connector JARs from the table in `$FLINK_CDC_HOME/lib`. + +Place the MySQL JDBC driver from the table in `$FLINK_HOME/lib` before starting Flink. +Save the Hadoop JAR as `$FLINK_HOME/opt/flink-shaded-hadoop.jar` and the Paimon SQL connector +as `$FLINK_HOME/opt/paimon-sql-connector.jar`. +The commands below add Hadoop with `--jar` when submitting the pipeline and verification queries. +If this Hadoop JAR is already in `$FLINK_HOME/lib`, move it to the `opt` path above before starting the cluster. +Loading it as a job dependency avoids conflicts with the libraries used by Flink's own CLI. + +In `$FLINK_HOME/conf/config.yaml`, add or update the following settings. +Merge them into existing `execution`, `jobmanager`, and `taskmanager` sections if those sections already exist: + +```yaml +execution: + checkpointing: + interval: 3s +jobmanager: + memory: + process: + size: 2048m + jvm-metaspace: + size: 512m +taskmanager: + memory: + process: + size: 2048m + jvm-metaspace: + size: 512m + numberOfTaskSlots: 4 +``` + +Checkpointing makes Paimon commits visible to readers. +Four task slots allow the ingestion pipeline and verification queries to run together. +These memory settings accommodate the verification queries and savepoint restore in this walkthrough. +Start the cluster: + +```shell +"$FLINK_HOME/bin/start-cluster.sh" +``` + +The [Flink Web UI](http://localhost:8081/) should show an available TaskManager. + +## Prepare the source tables + +Save the following content as `compose.yaml` in a new working directory: + +```yaml +services: + mysql: + image: mysql:8.0 + platform: linux/amd64 + ports: + - "3306:3306" + environment: + MYSQL_ROOT_PASSWORD: tutorial + MYSQL_ROOT_HOST: "%" + command: + - --server-id=1 + - --log-bin=mysql-bin + - --binlog-format=ROW + - --binlog-row-image=FULL + - --default-time-zone=+00:00 + healthcheck: + test: ["CMD", "mysqladmin", "ping", "-h", "127.0.0.1", "-uroot", "-ptutorial"] + interval: 2s + timeout: 5s + retries: 30 +``` + +Start MySQL and open its client: + +```shell +docker compose up -d --wait +docker compose exec mysql mysql -uroot -ptutorial +``` + +Create one business table, one temporary table, and one historical table: + +```sql +CREATE DATABASE inventory; +USE inventory; + +CREATE TABLE customers ( + id BIGINT NOT NULL, + name VARCHAR(100), + PRIMARY KEY (id) +); +INSERT INTO customers VALUES (101, 'Alice'); + +CREATE TABLE customers_tmp LIKE customers; +INSERT INTO customers_tmp VALUES (901, 'Temporary customer'); + +CREATE TABLE orders_history ( + id BIGINT NOT NULL, + total_price DECIMAL(10, 2), + PRIMARY KEY (id) +); +INSERT INTO orders_history VALUES (201, 100.00), (202, 200.00); +``` + +Keep this client open for subsequent changes. Run shell commands in another terminal. + +## Select tables and exclude temporary data + +Save the following configuration as `tables.yaml`: + +```yaml +source: + type: mysql + hostname: localhost + port: 3306 + username: root + password: tutorial + tables: inventory.\.* + tables.exclude: inventory.customers_tmp,inventory.products_tmp,inventory.orders_history + server-id: 5400-5404 + server-time-zone: UTC + scan.binlog.newly-added-table.enabled: true + +sink: + type: paimon + catalog.properties.metastore: filesystem + catalog.properties.warehouse: file:///tmp/flink-cdc-table-selection/warehouse + +pipeline: + name: Table selection and new tables + parallelism: 2 + schema.change.behavior: evolve +``` + +The `tables` expression selects tables in `inventory`. +The `tables.exclude` list excludes the two temporary-table names and the existing `orders_history` table. +An exclusion takes precedence when a table matches both expressions. + +The warehouse and savepoint paths in this example use the local filesystem. +For a distributed cluster, use shared storage accessible from all Flink processes. + +Submit the pipeline and record the Job ID: + +```shell +"$FLINK_CDC_HOME/bin/flink-cdc.sh" tables.yaml \ + --jar "$FLINK_HOME/opt/flink-shaded-hadoop.jar" +``` + +Wait for the initial snapshot and a completed checkpoint. Open the Flink SQL Client: + +```shell +"$FLINK_HOME/bin/sql-client.sh" \ + --jar "$FLINK_HOME/opt/paimon-sql-connector.jar" \ + --jar "$FLINK_HOME/opt/flink-shaded-hadoop.jar" +``` + +Create the catalog and verify the captured tables: + +```sql +SET 'execution.runtime-mode' = 'batch'; +SET 'sql-client.execution.result-mode' = 'tableau'; +SET 'parallelism.default' = '1'; + +CREATE CATALOG lake WITH ( + 'type' = 'paimon', + 'metastore' = 'filesystem', + 'warehouse' = 'file:///tmp/flink-cdc-table-selection/warehouse', + 'cache-enabled' = 'false' +); +USE CATALOG lake; + +SHOW TABLES FROM inventory; +SELECT * FROM inventory.customers ORDER BY id; +``` + +Only `customers` should appear in the table list, containing `(101, Alice)`. +Neither `customers_tmp` nor `orders_history` should exist in the target catalog. + +## Capture a newly created table without restarting + +First, update the existing customer in MySQL: + +```sql +UPDATE inventory.customers SET name = 'Alicia' WHERE id = 101; +``` + +Run the following query in the SQL Client until the name becomes `Alicia`: + +```sql +SELECT * FROM inventory.customers ORDER BY id; +``` + +This confirms that the job is consuming incremental changes before you create the new table. + +In MySQL, create and populate a business table and another temporary table: + +```sql +CREATE TABLE inventory.products ( + id BIGINT NOT NULL, + name VARCHAR(100), + price DECIMAL(10, 2), + PRIMARY KEY (id) +); +INSERT INTO inventory.products VALUES (301, 'Book', 30.00); + +CREATE TABLE inventory.products_tmp LIKE inventory.products; +INSERT INTO inventory.products_tmp VALUES (999, 'Temporary product', 1.00); +``` + +After the changes and a subsequent checkpoint complete, query Paimon: + +```sql +SHOW TABLES FROM inventory; +SELECT * FROM inventory.products ORDER BY id; +``` + +The table list should contain `customers` and `products`. +The product query should return `(301, Book, 30.00)`. +The job keeps the same Job ID, and both temporary tables remain excluded. + +The new table must match `tables` and must not match `tables.exclude`. +This mode reads the new table's changes from binlog; use the restore procedure below when historical rows need a snapshot. + +## Include a table with historical data + +### Stop the job with a savepoint + +Replace `<job-id>` with the running ingestion Job ID: + +```shell +"$FLINK_HOME/bin/flink" stop \ + --savepointPath file:///tmp/flink-cdc-table-selection/savepoints \ + <job-id> +``` + +Wait for the command to finish and record the complete savepoint path printed by Flink. +Keep the savepoint, the Paimon warehouse, and the MySQL binlog needed for recovery. + +While the ingestion job is stopped, run the following changes in MySQL: + +```sql +UPDATE inventory.customers SET name = 'Alice Smith' WHERE id = 101; +INSERT INTO inventory.orders_history VALUES (203, 300.00); +``` + +The customer update will verify recovery of a previously captured table. +The new order will be included with the other historical orders when that table is first captured. + +### Update the source configuration and restore + +Replace the contents of `tables.yaml` with: + +```yaml +source: + type: mysql + hostname: localhost + port: 3306 + username: root + password: tutorial + tables: inventory.\.* + tables.exclude: inventory.customers_tmp,inventory.products_tmp + server-id: 5400-5404 + server-time-zone: UTC + scan.binlog.newly-added-table.enabled: false + scan.newly-added-table.enabled: true + +sink: + type: paimon + catalog.properties.metastore: filesystem + catalog.properties.warehouse: file:///tmp/flink-cdc-table-selection/warehouse + +pipeline: + name: Table selection and new tables + parallelism: 2 + schema.change.behavior: evolve +``` + +The historical table is no longer excluded. +The binlog-only option is disabled, and snapshot scanning for newly included tables is enabled. + +Keep the sink, warehouse, pipeline topology, and parallelism unchanged. +For this workflow, leave `scan.incremental.snapshot.metadata.release.enabled` at its default value of `false`; +a savepoint from a job that has already released this metadata cannot be used to enable newly added table scanning. + +Replace the placeholder below with the complete savepoint path returned by the stop command: + +```shell +export SAVEPOINT_PATH=/path/to/savepoint + +"$FLINK_CDC_HOME/bin/flink-cdc.sh" tables.yaml \ + --from-savepoint "$SAVEPOINT_PATH" \ + --jar "$FLINK_HOME/opt/flink-shaded-hadoop.jar" +``` + +Record the new Job ID. Wait for the historical-table snapshot and a successful checkpoint, then query Paimon: + +```sql +SHOW TABLES FROM inventory; +SELECT * FROM inventory.customers ORDER BY id; +SELECT * FROM inventory.orders_history ORDER BY id; +``` + +The target catalog should now contain `customers`, `products`, and `orders_history`. +The customer should be `(101, Alice Smith)`, proving that changes made while the job was stopped were recovered. + +The historical table should contain all three rows: + +| id | total_price | +|----|-------------| +| 201 | 100.00 | +| 202 | 200.00 | +| 203 | 300.00 | + +The temporary tables should still be absent. + +## Verify subsequent changes to the added table + +Run the following statements in MySQL: + +```sql +UPDATE inventory.orders_history SET total_price = 120.00 WHERE id = 201; +DELETE FROM inventory.orders_history WHERE id = 202; +INSERT INTO inventory.orders_history VALUES (204, 400.00); +``` + +After a subsequent checkpoint completes, query the historical table again: + +```sql +SELECT * FROM inventory.orders_history ORDER BY id; +``` + +The result should be: + +| id | total_price | +|----|-------------| +| 201 | 120.00 | +| 203 | 300.00 | +| 204 | 400.00 | + +The added table now participates in normal incremental ingestion, together with the previously captured tables. +For source option details, see the [MySQL Pipeline connector](../connectors/pipeline-connectors/mysql.md). + +## Clean up + +Use the restored Job ID to cancel the job, then stop the tutorial services: + +```shell +"$FLINK_HOME/bin/flink" cancel <restored-job-id> +"$FLINK_HOME/bin/stop-cluster.sh" +docker compose down +``` + +The warehouse and savepoints remain under `/tmp/flink-cdc-table-selection`. +Use empty tutorial storage when starting this example again from scratch. Review Comment: ditto. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
