MaxCompute connector

Syntax

CREATE TEMPORARY TABLE odps_source(
  id INT,
  user_name VARCHAR,
  content VARCHAR
) WITH (
  'connector' = 'odps',
  'endpoint' = '<yourEndpoint>',
  'project' = '<yourProjectName>',
  'schemaName' = '<yourSchemaName>',
  'tableName' = '<yourTableName>',
  'accessId' = '${secret_values.ak_id}',
  'accessKey' = '${secret_values.ak_secret}',
  'partition' = 'ds=2018****'
);

Connector options

Syntax

source:
  type: xxx

sink:
  type: maxcompute
  name: MaxComputeSink
  access-id: ${your_accessId}
  access-key: ${your_accessKey}
  endpoint: ${your_maxcompute_endpoint}
  project: ${your_project}
  buckets-num: 8

Configuration options

Table location mappings

When the connector auto-creates tables in MaxCompute, locations are mapped as follows:

Data type mappings

For an example on how to enable EVOLVE mode, see Enable EVOLVE mode.

Write to a partitioned table

The create_time field of the TIMESTAMP type in the source table is converted to the DATE type and used as the partition key of the MaxCompute 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
    # (Optional) Synchronize the data of tables created in the incremental phase.
    scan.binlog.newly-added-table.enabled: true
    # (Optional) Synchronize table comments and column comments.
    include-comments.enabled: true
    # (Optional) Assign unbounded chunks first to prevent possible TaskManager out-of-memory errors.
    scan.incremental.snapshot.unbounded-chunk-first.enabled: true
    # (Optional) Enable parse filtering to accelerate reads.
    scan.only.deserialize.captured.tables.changelog.enabled: true

  sink:
    type: maxcompute
    name: MaxComputeSink
    access-id: ${your_accessId}
    access-key: ${your_accessKey}
    endpoint: ${your_maxcompute_endpoint}
    project: ${your_project}
    buckets-num: 8
 
  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 MaxCompute Pipeline

Synchronize a single table

Synchronize tables that require high real-time performance and stability in single-table synchronization jobs. This avoids the risks of complexity caused by multi-table synchronization. The following example shows the configuration:

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
    # (Optional) Synchronize table comments and column comments.
    include-comments.enabled: true
    # (Optional) Assign unbounded chunks first to prevent possible TaskManager out-of-memory errors.
    scan.incremental.snapshot.unbounded-chunk-first.enabled: true
    # (Optional) Enable parse filtering to accelerate reads.
    scan.only.deserialize.captured.tables.changelog.enabled: true


  sink:
    type: maxcompute
    name: MaxComputeSink
    access-id: ${your_accessId}
    access-key: ${your_accessKey}
    endpoint: ${your_maxcompute_endpoint}
    project: ${your_project}
    buckets-num: 8

  pipeline:
    name: MySQL to MaxCompute Pipeline

Synchronize an entire database

Whole-database synchronization reduces the number of synchronization jobs that you need to configure and run, which lowers O&M costs. The following example shows the configuration:

source:
    type: mysql
    name: MySQL Source
    hostname: <yourHostname>
    port: 3306
    username: flink
    password: ${secret_values.password}
    tables: test_db.\.*
    server-id: 5401-5499
    # (Optional) Synchronize the data of tables created in the incremental phase.
    scan.binlog.newly-added-table.enabled: true
    # (Optional) Synchronize table comments and column comments.
    include-comments.enabled: true
    # (Optional) Assign unbounded chunks first to prevent possible TaskManager out-of-memory errors.
    scan.incremental.snapshot.unbounded-chunk-first.enabled: true
    # (Optional) Enable parse filtering to accelerate reads.
    scan.only.deserialize.captured.tables.changelog.enabled: true


  sink:
    type: maxcompute
    name: MaxComputeSink
    access-id: ${your_accessId}
    access-key: ${your_accessKey}
    endpoint: ${your_maxcompute_endpoint}
    project: ${your_project}
    buckets-num: 8

  pipeline:
    name: MySQL to MaxCompute Pipeline

Synchronize an entire database and rename databases and tables

Based on whole-database synchronization, you can add the route module to uniformly rename databases and tables. The following example shows the configuration:

source:
    type: mysql
    name: MySQL Source
    hostname: <yourHostname>
    port: 3306
    username: flink
    password: ${secret_values.password}
    tables: test_db.\.*
    server-id: 5401-5499
    # (Optional) Synchronize the data of tables created in the incremental phase.
    scan.binlog.newly-added-table.enabled: true
    # (Optional) Synchronize table comments and column comments.
    include-comments.enabled: true
    # (Optional) Assign unbounded chunks first to prevent possible TaskManager out-of-memory errors.
    scan.incremental.snapshot.unbounded-chunk-first.enabled: true
    # (Optional) Enable parse filtering to accelerate reads.
    scan.only.deserialize.captured.tables.changelog.enabled: true


  sink:
    type: maxcompute
    name: MaxComputeSink
    access-id: ${your_accessId}
    access-key: ${your_accessKey}
    endpoint: ${your_maxcompute_endpoint}
    project: ${your_project}
    buckets-num: 8
  
  route:
    # Synchronize all tables in the MySQL test_db database to the MaxCompute test_db2 database and retain the table names.
    - source-table: test_db.\.*
      sink-table: test_db2.<>
      replace-symbol: <>

  pipeline:
    name: MySQL to MaxCompute Pipeline

Merge sharded tables

To write the data of multiple sharded tables into one destination table, use the following configuration template:

source:
    type: mysql
    name: MySQL Source
    hostname: <yourHostname>
    port: 3306
    username: flink
    password: ${secret_values.password}
    tables: test_db.user\.*
    server-id: 5401-5499
    # (Optional) Synchronize the data of tables created in the incremental phase.
    scan.binlog.newly-added-table.enabled: true
    # (Optional) Synchronize table comments and column comments.
    include-comments.enabled: true
    # (Optional) Assign unbounded chunks first to prevent possible TaskManager out-of-memory errors.
    scan.incremental.snapshot.unbounded-chunk-first.enabled: true
    # (Optional) Enable parse filtering to accelerate reads.
    scan.only.deserialize.captured.tables.changelog.enabled: true


  sink:
    type: maxcompute
    name: MaxComputeSink
    access-id: ${your_accessId}
    access-key: ${your_accessKey}
    endpoint: ${your_maxcompute_endpoint}
    project: ${your_project}
    buckets-num: 8
  
  route:
    # All sharded tables in MySQL test_db are merged into one MaxCompute table named test_db.user.
    - source-table: test_db.user\.*
      sink-table: test_db.user

  pipeline:
    name: MySQL to MaxCompute Pipeline

Synchronize existing tables after a restart

To add existing tables for 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
    # When the job restarts, check the new tables captured by the tables parameter and take a snapshot.
    # Note: Use this parameter together with scan.startup.mode: initial.
    scan.newly-added-table.enabled: true
    # (Optional) Synchronize the data of tables created in the incremental phase.
    scan.binlog.newly-added-table.enabled: true
    # (Optional) Synchronize table comments and column comments.
    include-comments.enabled: true
    # (Optional) Assign unbounded chunks first to prevent possible TaskManager out-of-memory errors.
    scan.incremental.snapshot.unbounded-chunk-first.enabled: true
    # (Optional) Enable parse filtering to accelerate reads.
    scan.only.deserialize.captured.tables.changelog.enabled: true


  sink:
    type: maxcompute
    name: MaxComputeSink
    access-id: ${your_accessId}
    access-key: ${your_accessKey}
    endpoint: ${your_maxcompute_endpoint}
    project: ${your_project}
    buckets-num: 8

  pipeline:
    name: MySQL to MaxCompute Pipeline

Exclude specific tables during whole-database synchronization

To exclude specific tables, such as test tables or tables that cause continuous failovers, during whole-database synchronization, use the following configuration:

source:
    type: mysql
    name: MySQL Source
    hostname: <yourHostname>
    port: 3306
    username: flink
    password: ${secret_values.password}
    tables: test_db.\.*
    # Tables that match this regular expression are not synchronized.
    tables.exclude: test_db.table1
    server-id: 5401-5499
    # (Optional) Synchronize the data of tables created in the incremental phase.
    scan.binlog.newly-added-table.enabled: true
    # (Optional) Synchronize table comments and column comments.
    include-comments.enabled: true
    # (Optional) Assign unbounded chunks first to prevent possible TaskManager out-of-memory errors.
    scan.incremental.snapshot.unbounded-chunk-first.enabled: true
    # (Optional) Enable parse filtering to accelerate reads.
    scan.only.deserialize.captured.tables.changelog.enabled: true

  sink:
    type: maxcompute
    name: MaxComputeSink
    access-id: ${your_accessId}
    access-key: ${your_accessKey}
    endpoint: ${your_maxcompute_endpoint}
    project: ${your_project}
    buckets-num: 8

  pipeline:
    name: MySQL to MaxCompute Pipeline

Enable EVOLVE mode

EVOLVE mode keeps the schema of the destination table strictly consistent with the schema of the source table, including operations such as dropping fields and dropping tables. In this mode, if the destination table cannot apply all schema change events, the job may fail over and cannot automatically recover. The following example shows the job configuration:

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
    # (Optional) Synchronize the data of tables created in the incremental phase.
    scan.binlog.newly-added-table.enabled: true
    # (Optional) Synchronize table comments and column comments.
    include-comments.enabled: true
    # (Optional) Assign unbounded chunks first to prevent possible TaskManager out-of-memory errors.
    scan.incremental.snapshot.unbounded-chunk-first.enabled: true
    # (Optional) Enable parse filtering to accelerate reads.
    scan.only.deserialize.captured.tables.changelog.enabled: true

  sink:
    type: maxcompute
    name: MaxComputeSink
    access-id: ${your_accessId}
    access-key: ${your_accessKey}
    endpoint: ${your_maxcompute_endpoint}
    project: ${your_project}
    buckets-num: 8

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

SQL API

CREATE TEMPORARY TABLE odps_source (
  cid VARCHAR,
  rt DOUBLE
) WITH (
  'connector' = 'odps',
  'endpoint' = '<yourEndpointName>',
  'project' = '<yourProjectName>',
  'tableName' = '<yourTableName>',
  'accessId' = '${secret_values.ak_id}',
  'accessKey' = '${secret_values.ak_secret}',
  'partition' = 'ds=201809*'
);

CREATE TEMPORARY TABLE blackhole_sink (
  cid VARCHAR,
  invoke_count BIGINT
) WITH (
  'connector' = 'blackhole'
);

INSERT INTO blackhole_sink
SELECT
   cid,
   COUNT(*) AS invoke_count
FROM odps_source GROUP BY cid;

CREATE TEMPORARY TABLE odps_source (
  cid VARCHAR,
  rt DOUBLE
) WITH (
  'connector' = 'odps',
  'endpoint' = '<yourEndpointName>',
  'project' = '<yourProjectName>',
  'tableName' = '<yourTableName>',
  'accessId' = '${secret_values.ak_id}',
  'accessKey' = '${secret_values.ak_secret}',
  'startPartition' = 'yyyy=2018,MM=09,dd=05' -- Start reading from the 20180905 partition.
);

CREATE TEMPORARY TABLE blackhole_sink (
  cid VARCHAR,
  invoke_count BIGINT
) WITH (
  'connector' = 'blackhole'
);

INSERT INTO blackhole_sink
SELECT cid, COUNT(*) AS invoke_count
FROM odps_source GROUP BY cid;

CREATE TEMPORARY TABLE datagen_source (
  id INT,
  len INT,
  content VARCHAR
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE odps_sink (
  id INT,
  len INT,
  content VARCHAR
) WITH (
  'connector' = 'odps',
  'endpoint' = '<yourEndpoint>',
  'project' = '<yourProjectName>',
  'tableName' = '<yourTableName>',
  'accessId' = '${secret_values.ak_id}',
  'accessKey' = '${secret_values.ak_secret}',
  'partition' = 'ds=20180905' -- Write to partition 20180905.
);

INSERT INTO odps_sink
SELECT
  id, len, content
FROM datagen_source;

CREATE TEMPORARY TABLE datagen_source (
  id INT,
  len INT,
  content VARCHAR,
  c TIMESTAMP
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE odps_sink (
  id  INT,
  len INT,
  content VARCHAR,
  ds VARCHAR -- Dynamic partition column.
) WITH (
  'connector' = 'odps',
  'endpoint' = '<yourEndpoint>',
  'project' = '<yourProjectName>',
  'tableName' = '<yourTableName>',
  'accessId' = '${secret_values.ak_id}',
  'accessKey' = '${secret_values.ak_secret}',
  'partition' = 'ds' -- Omit the value; data is routed to partitions based on the ds field.
);

INSERT INTO odps_sink
SELECT
   id,
   len,
   content,
   DATE_FORMAT(c, 'yyMMdd') as ds
FROM datagen_source;

CREATE TEMPORARY TABLE datagen_source (
  k INT,
  v VARCHAR
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE odps_dim (
  k INT,
  v VARCHAR,
  PRIMARY KEY (k) NOT ENFORCED  -- Specify the primary key.
) WITH (
  'connector' = 'odps',
  'endpoint' = '<yourEndpoint>',
  'project' = '<yourProjectName>',
  'tableName' = '<yourTableName>',
  'accessId' = '${secret_values.ak_id}',
  'accessKey' = '${secret_values.ak_secret}',
  'partition' = 'ds=20180905',
  'cache' = 'ALL'
);

CREATE TEMPORARY TABLE blackhole_sink (
  k VARCHAR,
  v1 VARCHAR,
  v2 VARCHAR
) WITH (
  'connector' = 'blackhole'
);

INSERT INTO blackhole_sink
SELECT k, s.v, d.v
FROM datagen_source AS s
INNER JOIN odps_dim FOR SYSTEM_TIME AS OF PROCTIME() AS d ON s.k = d.k;

CREATE TEMPORARY TABLE datagen_source (
  k INT,
  v VARCHAR
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE odps_dim (
  k INT,
  v VARCHAR
  -- No primary key needed for multi-value lookups.
) WITH (
  'connector' = 'odps',
  'endpoint' = '<yourEndpoint>',
  'project' = '<yourProjectName>',
  'tableName' = '<yourTableName>',
  'accessId' = '${secret_values.ak_id}',
  'accessKey' = '${secret_values.ak_secret}',
  'partition' = 'ds=20180905',
  'cache' = 'ALL'
);

CREATE TEMPORARY TABLE blackhole_sink (
  k VARCHAR,
  v1 VARCHAR,
  v2 VARCHAR
) WITH (
  'connector' = 'blackhole'
);

INSERT INTO blackhole_sink
SELECT k, s.v, d.v
FROM datagen_source AS s
INNER JOIN odps_dim FOR SYSTEM_TIME AS OF PROCTIME() AS d ON s.k = d.k;

DataStream API

Declare the MaxCompute table using SQL, then access it via the Table API or DataStream API.

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
tEnv.executeSql(String.join(
    "\n",
    "CREATE TEMPORARY TABLE IF NOT EXISTS odps_source (",
    "  cid VARCHAR,",
    "  rt DOUBLE",
    ") WITH (",
    "  'connector' = 'odps',",
    "  'endpoint' = '<yourEndpointName>',",
    "  'project' = '<yourProjectName>',",
    "  'tableName' = '<yourTableName>',",
    "  'accessId' = '<yourAccessId>',",
    "  'accessKey' = '<yourAccessPassword>',",
    "  'partition' = 'ds=201809*'",
    ")");
DataStream<Row> source = tEnv.toDataStream(tEnv.from("odps_source"));
source.print();
env.execute("odps source");

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
tEnv.executeSql(String.join(
    "\n",
    "CREATE TEMPORARY TABLE IF NOT EXISTS odps_sink (",
    "  cid VARCHAR,",
    "  rt DOUBLE",
    ") WITH (",
    "  'connector' = 'odps',",
    "  'endpoint' = '<yourEndpointName>',",
    "  'project' = '<yourProjectName>',",
    "  'tableName' = '<yourTableName>',",
    "  'accessId' = '<yourAccessId>',",
    "  'accessKey' = '<yourAccessPassword>',",
    "  'partition' = 'ds=20180905'",
    ")");
DataStream<Row> data = env.fromElements(
    Row.of("id0", 3.),
    Row.of("id1", 4.));
tEnv.fromDataStream(data).insertInto("odps_sink").execute();

<dependency>
    <groupId>com.alibaba.ververica</groupId>
    <artifactId>ververica-connector-odps</artifactId>
    <version>${vvr-version}</version>
</dependency>