Realtime Compute for Apache Flink:SelectDB

General

Source table

Sink table

Write mode affects delivery guarantees and flush behavior. Choose based on your consistency requirements:

**sink.properties.* examples:**

CSV format:

'sink.properties.column_separator' = ','
-- If values may contain commas, use a non-printable separator:
-- 'sink.properties.column_separator' = '\x01'

JSON format:

'sink.properties.format'            = 'json',
'sink.properties.read_json_by_line' = 'true'
-- Alternatively: 'sink.properties.strip_outer_array' = 'true'

Dimension table

Single-table synchronization

Synchronize a single MySQL table to SelectDB. The destination table is created automatically if it does not exist.

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: ${secret_values.mysql_username}
  password: ${secret_values.mysql_password}
  tables: test_db.test_source_table
  server-id: 5401-5499
  # (Optional) Synchronize the existing full data first, and then the incremental data.
  scan.startup.mode: initial

sink:
  type: doris
  name: SelectDB Sink
  fenodes: selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080
  jdbc-url: jdbc:mysql://selectdb-cn-****.selectdbfe.rds.aliyuncs.com:9030
  username: ${secret_values.selectdb_username}
  password: ${secret_values.selectdb_password}

pipeline:
  name: MySQL to SelectDB Pipeline

Full database synchronization

Synchronize all tables in a MySQL database to SelectDB. The downstream database and table names are the same as the upstream ones. Destination tables are created automatically if they do not exist.

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: ${secret_values.mysql_username}
  password: ${secret_values.mysql_password}
  # Matches all tables in the test_db database.
  tables: test_db.\.*
  server-id: 5401-5499
  scan.startup.mode: initial
  # (Optional) Synchronize tables that are created during the incremental phase without restarting the job.
  scan.binlog.newly-added-table.enabled: true

sink:
  type: doris
  name: SelectDB Sink
  fenodes: selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080
  jdbc-url: jdbc:mysql://selectdb-cn-****.selectdbfe.rds.aliyuncs.com:9030
  username: ${secret_values.selectdb_username}
  password: ${secret_values.selectdb_password}
  # (Optional) The flush interval for batch writing. Lower this value for small workloads so that data is not held in the buffer for too long.
  sink.buffer-flush.interval: 10s

pipeline:
  name: MySQL to SelectDB Pipeline

Excluding tables from full database synchronization

Skip the tables that you do not want to synchronize, such as temporary tables and log tables. Without a route module, data is written to the destination table that has the same name as the source table.

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: ${secret_values.mysql_username}
  password: ${secret_values.mysql_password}
  tables: test_db.\.*
  # Tables that match this regular expression are not synchronized. Separate multiple regular expressions with commas (,).
  tables.exclude: test_db.tmp_\.*, test_db.log_\.*
  server-id: 5401-5499
  scan.startup.mode: initial

sink:
  type: doris
  name: SelectDB Sink
  fenodes: selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080
  jdbc-url: jdbc:mysql://selectdb-cn-****.selectdbfe.rds.aliyuncs.com:9030
  username: ${secret_values.selectdb_username}
  password: ${secret_values.selectdb_password}

pipeline:
  name: MySQL to SelectDB Pipeline

Synchronizing to specified databases and tables

Use the route module to rename destinations when the downstream database or table names differ from the upstream ones. The following example synchronizes the tables in test_db to the ods_db database and adds the ods_ prefix to each table name. For example, test_db.orders is synchronized to ods_db.ods_orders.

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: ${secret_values.mysql_username}
  password: ${secret_values.mysql_password}
  tables: test_db.\.*
  server-id: 5401-5499
  scan.startup.mode: initial

sink:
  type: doris
  name: SelectDB Sink
  fenodes: selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080
  jdbc-url: jdbc:mysql://selectdb-cn-****.selectdbfe.rds.aliyuncs.com:9030
  username: ${secret_values.selectdb_username}
  password: ${secret_values.selectdb_password}

route:
  # <> is a placeholder that is replaced with the matched source table name.
  - source-table: test_db.\.*
    sink-table: ods_db.ods_<>
    replace-symbol: <>

pipeline:
  name: MySQL to SelectDB Pipeline

Merging sharded tables

Merge multiple shard tables that share the same schema into a single SelectDB table. The following example uses the transform module to append a column that identifies the source database and table, and then uses this column with the business primary key as a composite primary key so that you can trace the origin of the merged data.

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: ${secret_values.mysql_username}
  password: ${secret_values.mysql_password}
  # Matches all shards, such as order_db_1.orders_1 and order_db_2.orders_2.
  tables: order_db_\d+.orders_\d+
  server-id: 5401-5499
  scan.startup.mode: initial

sink:
  type: doris
  name: SelectDB Sink
  fenodes: selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080
  jdbc-url: jdbc:mysql://selectdb-cn-****.selectdbfe.rds.aliyuncs.com:9030
  username: ${secret_values.selectdb_username}
  password: ${secret_values.selectdb_password}
  # Set this parameter to false when the downstream primary key differs from the upstream one. Otherwise, the rows that correspond to the old primary key are not deleted.
  sink.ignore.update-before: false

transform:
  - source-table: order_db_\d+.orders_\d+
    # Append the src_table column that identifies the source database and table.
    projection: "*, __schema_name__ || '.' || __table_name__ AS src_table"
    # The composite primary key must be the same as the key columns of the destination table.
    primary-keys: order_id, src_table
    description: Append the source identifier and set a composite primary key

route:
  # Merge all shards into a single table.
  - source-table: order_db_\d+.orders_\d+
    sink-table: dw_db.merged_orders

pipeline:
  name: MySQL sharding to SelectDB Pipeline

Filtering data and pruning columns

Use the transform module to filter data and prune columns. The filter condition is evaluated on the source table fields to decide whether a record is synchronized, and the projection decides which columns are written to the downstream table.

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: ${secret_values.mysql_username}
  password: ${secret_values.mysql_password}
  tables: test_db.test_source_table
  server-id: 5401-5499
  scan.startup.mode: initial

sink:
  type: doris
  name: SelectDB Sink
  fenodes: selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080
  jdbc-url: jdbc:mysql://selectdb-cn-****.selectdbfe.rds.aliyuncs.com:9030
  username: ${secret_values.selectdb_username}
  password: ${secret_values.selectdb_password}

transform:
  - source-table: test_db.test_source_table
    # Synchronize only the records whose order_status is PAID.
    filter: "order_status = 'PAID'"
    # Write only the following four columns.
    projection: order_id, customer_id, total_amount, created_at
    primary-keys: order_id
    description: Filter by order status and prune columns

pipeline:
  name: MySQL to SelectDB Pipeline

Full PostgreSQL database synchronization

Synchronize all tables in a PostgreSQL schema to SelectDB. Before you start, set wal_level to logical on the PostgreSQL instance and make sure that max_replication_slots and max_wal_senders have enough headroom. For more information, see Configure a PostgreSQL database.

source:
  type: postgres
  name: PostgreSQL Source
  hostname: <yourHostname>
  port: 5432
  username: ${secret_values.pg_username}
  password: ${secret_values.pg_password}
  # PostgreSQL table names use the database.schema.table format.
  tables: test_db.public.\.*
  slot.name: <yourSlotName>
  decoding.plugin.name: pgoutput
  scan.startup.mode: initial
  # (Optional) Use an existing publication instead of creating one when the job starts.
  debezium.publication.autocreate.mode: disabled
  debezium.publication.name: <yourPublicationName>

sink:
  type: doris
  name: SelectDB Sink
  fenodes: selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080
  jdbc-url: jdbc:mysql://selectdb-cn-****.selectdbfe.rds.aliyuncs.com:9030
  username: ${secret_values.selectdb_username}
  password: ${secret_values.selectdb_password}

pipeline:
  name: PostgreSQL to SelectDB Pipeline