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

Genel seçenekler

Kaynak seçenekleri

Artımlı kaynak seçenekleri

Artımlı kaynak, yeni bölümleri keşfetmek için MaxCompute'i aralıklı olarak yoklar. Yeni bir bölümü okumadan önce, o bölüme yapılan tüm veri yazma işlemlerinin tamamlanmış olması gerekir. Ayrıntılar için Yukarı ve aşağı akış depolama hakkında SSS içindeki "Artımlı bir kaynak veri yazılırken yeni bir bölüm algılarsa ne yapmalıyım?" bölümüne bakın.

Bölüm sıralaması: Kaynak, alfabetik sırası startPartition değerine eşit veya ondan büyük olan bölümleri okur. Örneğin, year=2023,month=10 alfabetik olarak year=2023,month=9 değerinden önce gelir; doğru sıralamayı sağlamak için ay değerlerini sıfırla doldurun (year=2023,month=9 yerine year=2023,month=09 kullanın).

Havuz seçenekleri

Boyut tablosu seçenekleri

Dağıtım başladığında boyut tablosu partition ile belirtilen bölümden tüm verileri yükler. partition seçeneği max_pt() işlevini destekler. Önbellek yeniden yüklendiğinde en son bölüm tekrar okunur. İki bölümden veri yüklemek için partition değerini max_two_pt() olarak ayarlayın.

Boyut tabloları cache=ALL gerektirir . Birleşim düğümü belleğini uzak tablo verisi boyutunun en az dört katına çıkarın. Büyük boyut tabloları için verileri eşit şekilde dağıtmak amacıyla SHUFFLE_HASH ipucunu kullanın. Sık Java Sanal Makinesi (JVM) çöp toplama (GC) işlemlerine neden olan çok büyük tablolar için LRU (En Az Son Kullanılan) önbellek ilkesiyle bir anahtar-değer boyut tablosuna dönüştürün; örneğin, bir ApsaraDB for HBase boyut tablosu.

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 (varsayılan)

LENIENT modu aşağıdaki şema değişikliklerini işler:

• Boş bırakılabilir bir sütun ekleme: Sütun, hedef tablonun şemasının sonuna eklenir ve yeni sütunun verileri otomatik olarak senkronize edilir.

• Boş bırakılabilir bir sütunu kaldırma: Sütun hedef tablodan kaldırılmaz. Sütunun verileri otomatik olarak NULL değerleriyle doldurulur.

• Boş bırakılamaz bir sütun ekleme: Sütun, hedef tablonun şemasının sonuna eklenir ve yeni sütunun verileri otomatik olarak senkronize edilir. Yeni sütun varsayılan olarak boş bırakılabilirdir. Sütun eklenmeden önce oluşturulan veriler NULL olarak ayarlanır.

• Bir sütunu yeniden adlandırma: Olay, bir sütun ekleme ve bir sütun kaldırma olarak işlenir. Yeniden adlandırılan sütun hedef tablonun sonuna eklenir ve orijinal sütunun verileri otomatik olarak NULL değerleriyle doldurulur. Örneğin, col_a col_b olarak yeniden adlandırılırsa col_b hedef tablonun sonuna eklenir ve col_a'nın verileri otomatik olarak NULL değerleriyle doldurulur.

• Bir sütun veri türünü değiştirme: Sütunun veri türü hedef tabloda değiştirilir.

• Aşağıdaki şema değişiklikleri desteklenmez:

  • Bir sütunu NULLABLE'dan NOT NULL'a değiştirme.

EVOLVE

EVOLVE modu aşağıdaki şema değişikliklerini işler:

• Boş bırakılabilir bir sütun ekleme: desteklenir.

• Boş bırakılabilir bir sütunu kaldırma: desteklenmez.

• Boş bırakılamaz bir sütun ekleme: Hedef tabloya boş bırakılabilir bir sütun eklenir.

• Bir sütunu yeniden adlandırma: desteklenir. Hedef tablodaki orijinal sütun yeniden adlandırılır.

• Bir sütun veri türünü değiştirme: Sütunun veri türü hedef tabloda değiştirilir.

• Aşağıdaki şema değişiklikleri desteklenmez:

  • Bir sütunu NULLABLE'dan NOT NULL'a değiştirme.

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

Maven bağımlılığı

MaxCompute DataStream bağlayıcısını projenize ekleyin. Tüm sürümler Maven merkezi deposunda mevcuttur.

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