Realtime Compute for Apache Flink:Hologres YAML connector for data ingestion

Reuse an existing Catalog

Starting from VVR 11.5, you can directly reference a built-in Hologres Catalog created on the Data Management page in a Flink CDC data ingestion job. This reduces the need to manually specify connection properties.

sink:
  type: hologres
  using.built-in-catalog: my_holo_catalog

Data ingestion jobs can automatically reuse the following Hologres Catalog parameters:

• endpoint

• username

• password

• dbname

To override these automatically reused parameters, you can explicitly specify the corresponding YAML parameters, which have a higher priority.

Data type mapping

You can use the parameter sink.type-normalize-strategy to set the strategy for converting upstream data to Hologres types.

LENIENT (default)

In LENIENT mode, schema changes are handled as follows:

• Add a nullable column: The corresponding column is automatically added to the end of the sink table, and its data is synchronized.

• Delete a nullable column: The column is not deleted from the sink table. Instead, the column is automatically populated with NULL values.

• Add a non-nullable column: A corresponding nullable column is automatically added to the end of the sink table, and its data is synchronized. For existing rows, this new column is automatically populated with NULL.

• Rename a column: This operation is treated as deleting a column and adding a new one. A new column with the specified name is added to the end of the sink table, and the column with the original name is automatically populated with NULL values. For example, if col_a is renamed to col_b, a col_b column is added to the end of the sink table, and the col_a column is automatically populated with NULL values.

• Change a column's data type: Not supported. As Hologres does not support changing a column's data type, you must use the sink.type-normalize-strategy parameter.

• The following schema changes are not supported:

  • Changes to constraints, such as the primary key or index.

  • Deleting a non-nullable column.

  • Changing a column from NOT NULL to NULLABLE.

EVOLVE

In EVOLVE mode, schema changes are handled as follows:

• Add a nullable column: Supported.

• Delete a nullable column: Not supported.

• Add a non-nullable column: A new nullable column is added to the sink table.

• Rename a column: Supported. The original column is renamed in the sink table.

• Change a column's data type: Not supported. As Hologres does not support changing a column's data type, you must use the sink.type-normalize-strategy parameter.

• The following schema changes are not supported:

  • Changes to constraints, such as the primary key or index.

  • Deleting a non-nullable column.

  • Changing a column from NOT NULL to NULLABLE.

Type broadening

Use the sink.type-normalize-strategy parameter to configure type broadening.

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  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: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Map CDC data types to broader Hologres types.
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline

Writing to a partitioned table

Convert the create_time timestamp field to a date type and use it as the partition key for the Hologres table.

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  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: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Automatically create partitioned tables if they do not exist.
  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

Enabling EVOLVE mode

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  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: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Automatically create partitioned tables if they do not exist.
  createparttable: true

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

Single-table synchronization

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  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: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Map CDC data types to broader Hologres types.
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline

Full database synchronization

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

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Map CDC data types to broader Hologres types.
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline

Merging sharded tables

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

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Map CDC data types to broader Hologres types.  
  sink.type-normalize-strategy: BROADEN
  
route:
  # All sharded tables in the MySQL test_db database are merged into a single Hologres table named test_db.user.
  - source-table: test_db.user\.*
    sink-table: test_db.user

pipeline:
  name: MySQL to Hologres Pipeline

Synchronizing to a specified schema

In Hologres, a schema corresponds to a database in MySQL. You can specify the schema for the sink tables.

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

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Map CDC data types to broader Hologres types.
  sink.type-normalize-strategy: BROADEN
  
route:
  # Synchronize all tables from the MySQL test_db database to the Hologres test_db2 schema while keeping the original table names.
  - source-table: test_db.\.*
    sink-table: test_db2.<>
    replace-symbol: <>

pipeline:
  name: MySQL to Hologres Pipeline

Synchronizing new tables without restart

To synchronize newly added tables in real time while a job is running, set scan.binlog.newly-added-table.enable = true.

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

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Map CDC data types to broader Hologres types.
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline

Adding existing tables on restart

To include an existing table in the synchronization, set scan.newly-added-table.enabled to true and restart the job.

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499
  scan.startup.mode: initial
  # On restart, the job scans for new tables matching the `tables` parameter and performs a snapshot.
  # Note: This parameter must be used with scan.startup.mode: initial.
  scan.newly-added-table.enabled: true

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Map CDC data types to broader Hologres types.
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline

Excluding tables

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  # Excludes tables that match this regular expression.
  tables.exclude: test_db.table1
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Map CDC data types to broader Hologres types.
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline