Realtime Compute for Apache Flink:Hologres data ingestion

Required parameters

Optional parameters

Write mode

Use mutatetype to control how the sink handles rows with duplicate primary keys:

Example: A table has columns a (primary key), b, c, d. Only a and b are written. With INSERT_OR_UPDATE, a duplicate primary key updates only b. Columns c and d stay unchanged.

Delete strategy

Use sink.delete-strategy to control how the sink handles retraction messages:

SSL encryption

connection.ssl.mode accepts the following values:

verify-ca and verify-full require Hologres 2.1 or later. See Encryption in transit. When using either mode, also set connection.ssl.root-cert.location.

To set connection.ssl.root-cert.location, upload the CA certificate through File Management in the Realtime Compute console. Uploaded files go to /flink/usrlib. For example, if your certificate file is named certificate.crt, set this parameter to /flink/usrlib/certificate.crt. To download the CA certificate, see Encryption in transit — Download the CA certificate.

STANDARD (default)

Maps Flink CDC types to PostgreSQL-standard Hologres types.

BROADEN

Maps Flink CDC types to wider Hologres types to reduce schema mismatch errors. Use when write compatibility is more important than type precision.

ONLY_BIGINT_OR_TEXT

Maps all Flink CDC types to either BIGINT or text in Hologres. Use when maximum type compatibility is required.

LENIENT (default)

Example: Renaming col_a to col_b adds col_b to the end of the sink table and fills col_a with null.

EVOLVE

Quickstart: sync a single table

The simplest end-to-end configuration: sync one MySQL table to Hologres with wide type mapping.

source:
  type: mysql
  name: MySQL Source
  hostname: <your-hostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.test_source_table
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <your-endpoint>
  dbname: <your-database>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline

Sync all tables in a database

Set tables to a regex pattern to capture all tables in a database.

source:
  type: mysql
  name: MySQL Source
  hostname: <your-hostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db\..*
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <your-endpoint>
  dbname: <your-database>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline

Merge sharded tables

Use a route rule to merge multiple sharded tables from MySQL into a single Hologres table.

source:
  type: mysql
  name: MySQL Source
  hostname: <your-hostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.user\..*
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <your-endpoint>
  dbname: <your-database>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  sink.type-normalize-strategy: BROADEN

route:
  # Merge all sharded tables in MySQL test_db into one Hologres table: test_db.user.
  - source-table: test_db.user\..*
    sink-table: test_db.user

pipeline:
  name: MySQL to Hologres Pipeline

Write to partitioned tables

Convert the upstream timestamp field create_time to a DATE partition key using the transform module.

source:
  type: mysql
  name: MySQL Source
  hostname: <your-hostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.test_source_table
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <your-endpoint>
  dbname: <your-database>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Create missing child partitioned tables automatically.
  createparttable: true

transform:
  - source-table: test_db.test_source_table
    projection: \*, DATE_FORMAT(CAST(create_time AS TIMESTAMP), 'yyyy-MM-dd') as partition_key
    primary-keys: id, create_time, partition_key
    partition-keys: partition_key
    description: add partition key

pipeline:
  name: MySQL to Hologres Pipeline

Sync to a specific schema

A schema in Hologres corresponds to a database in MySQL. Use a route rule with replace-symbol to redirect all tables from the source database into a target Hologres schema.

source:
  type: mysql
  name: MySQL Source
  hostname: <your-hostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.user\..*
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <your-endpoint>
  dbname: <your-database>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  sink.type-normalize-strategy: BROADEN

route:
  # Sync all tables from MySQL test_db to Hologres schema test_db2, keeping table names unchanged.
  - source-table: test_db\..*
    sink-table: test_db2.<>
    replace-symbol: <>

pipeline:
  name: MySQL to Hologres Pipeline

Enable EVOLVE mode

Set schema.change.behavior: evolve at the pipeline level to enable schema evolution.

source:
  type: mysql
  name: MySQL Source
  hostname: <your-hostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.test_source_table
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <your-endpoint>
  dbname: <your-database>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  createparttable: true

pipeline:
  name: MySQL to Hologres Pipeline
  schema.change.behavior: evolve

Sync newly created tables without restarting

Set scan.binlog.newly-added-table.enabled: true to capture tables created after the job starts.

source:
  type: mysql
  name: MySQL Source
  hostname: <your-hostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db\..*
  server-id: 5401-5499
  # Capture data from new tables created while the job is running.
  scan.binlog.newly-added-table.enabled: true

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <your-endpoint>
  dbname: <your-database>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline

Add existing tables on restart

To add synchronization for existing tables, set scan.newly-added-table.enabled: true and restart the job.

source:
  type: mysql
  name: MySQL Source
  hostname: <your-hostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db\..*
  server-id: 5401-5499
  scan.startup.mode: initial
  # On restart, detect new tables in the tables parameter and run a full snapshot for them.
  # Requires scan.startup.mode: initial.
  scan.newly-added-table.enabled: true

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <your-endpoint>
  dbname: <your-database>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline

Exclude tables during full-database sync

Use tables.exclude to skip specific tables when syncing an entire database.

source:
  type: mysql
  name: MySQL Source
  hostname: <your-hostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db\..*
  # Tables matching this regex are not synchronized.
  tables.exclude: test_db.table1
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <your-endpoint>
  dbname: <your-database>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline