Realtime Compute for Apache Flink:MaxCompute connector

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****'
);

Opciones generales

Opciones de origen

Opciones de origen incremental

El origen incremental consulta MaxCompute de forma intermitente para detectar nuevas particiones. Antes de leer una partición nueva, deben haber concluido todas las escrituras de datos en ella. Para más detalles, consulte la sección "What do I do if an incremental source detects a new partition while data is still being written?" de FAQ about upstream and downstream storage.

Orden de las particiones: El origen lee las particiones cuyo orden alfabético sea mayor o igual que el valor de startPartition. Por ejemplo, year=2023,month=10 se ordena antes que year=2023,month=9 alfabéticamente. Por ello, rellene los valores de mes con ceros (use year=2023,month=09 en lugar de year=2023,month=9) para garantizar un orden correcto.

Opciones de destino

Opciones de tabla de dimensiones

Cuando se inicia un despliegue, la tabla de dimensiones carga todos los datos de la partición especificada por partition. La opción partition es compatible con la función max_pt(). Al recargar la caché, se vuelve a leer la partición más reciente. Establezca partition en max_two_pt() para cargar datos de dos particiones.

Las tablas de dimensiones requieren cache=ALL . Aumente la memoria del nodo de unión a como mínimo cuatro veces el tamaño de los datos de la tabla remota. Para tablas de dimensiones grandes, use la sugerencia SHUFFLE_HASH para distribuir los datos de manera uniforme. En tablas extremadamente grandes que provoquen recolecciones de basura (GC) frecuentes en la máquina virtual Java (JVM), conviértalas a una tabla de dimensiones de clave-valor con una política de caché LRU (Least Recently Used); por ejemplo, una tabla de dimensiones de ApsaraDB for HBase.

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

LENIENT (predeterminado)

El modo LENIENT gestiona los siguientes cambios de esquema:

• Agregar una columna anulable: la columna se agrega al final del esquema de la tabla de destino y sus datos se sincronizan automáticamente.

• Eliminar una columna anulable: la columna no se elimina de la tabla de destino. Sus datos se rellenan automáticamente con valores NULL.

• Agregar una columna no anulable: la columna se agrega al final del esquema de la tabla de destino y sus datos se sincronizan automáticamente. De forma predeterminada, la nueva columna es anulable. Los datos generados antes de agregarla se establecen en NULL.

• Cambiar el nombre de una columna: el evento se trata como la adición de una columna y la eliminación de otra. La columna renombrada se agrega al final de la tabla de destino y los datos de la columna original se rellenan automáticamente con valores NULL. Por ejemplo, si col_a se renombra a col_b, col_b se agrega al final de la tabla de destino y los datos de col_a se rellenan con valores NULL.

• Cambiar el tipo de datos de una columna: el tipo de datos de la columna se modifica en la tabla de destino.

• Los siguientes cambios de esquema no son compatibles:

  • Cambiar una columna de NULLABLE a NOT NULL.

EVOLVE

El modo EVOLVE gestiona los siguientes cambios de esquema:

• Agregar una columna anulable: compatible.

• Eliminar una columna anulable: no compatible.

• Agregar una columna no anulable: se agrega una columna anulable a la tabla de destino.

• Cambiar el nombre de una columna: compatible. La columna original en la tabla de destino se renombra.

• Cambiar el tipo de datos de una columna: el tipo de datos de la columna se modifica en la tabla de destino.

• Los siguientes cambios de esquema no son compatibles:

  • Cambiar una columna de NULLABLE a NOT NULL.

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

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

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

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

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

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

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

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

Dependencia de Maven

Agregue el conector de DataStream de MaxCompute a su proyecto. Todas las versiones están disponibles en el repositorio central de Maven.

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