数据湖构建：Flink CDC访问DLF

MySQL整库同步数据湖DLF

MySQL整库同步数据到DLF的CDC YAML作业如下所示：

source:
  type: mysql
  name: MySQL Source
  hostname: ${mysql.hostname}
  port: ${mysql.port}
  username: ${mysql.username}
  password: ${mysql.password}
  tables: mysql_test.\.*
  server-id: 8601-8604
  #（可选）同步增量阶段新创建的表的数据
  scan.binlog.newly-added-table.enabled: true
  #（可选）同步表注释和字段注释
  include-comments.enabled: true
  #（可选）优先分发无界的分片以避免可能出现的TaskManager OutOfMemory问题
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #（可选）开启解析过滤，加速读取
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  #（可选）提交用户名，建议为不同作业设置不同的提交用户以避免冲突
  commit.user: your_job_name
  #（可选）开启删除向量，提升读取性能
  table.properties.deletion-vectors.enabled: true

source:
  type: mysql
  name: MySQL Source
  hostname: ${mysql.hostname}
  port: ${mysql.port}
  username: ${mysql.username}
  password: ${mysql.password}
  tables: mysql_test.\.*
  server-id: 8601-8604
  #（可选）同步增量阶段新创建的表的数据
  scan.binlog.newly-added-table.enabled: true
  #（可选）同步表注释和字段注释
  include-comments.enabled: true
  #（可选）优先分发无界的分片以避免可能出现的TaskManager OutOfMemory问题
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #（可选）开启解析过滤，加速读取
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  #（可选）提交用户名，建议为不同作业设置不同的提交用户以避免冲突
  commit.user: your_job_name
  #（可选）开启删除向量，提升读取性能
  table.properties.deletion-vectors.enabled: true

transform:
  - source-table: mysql_test.tbl1
    #（可选）设置分区字段  
    partition-keys: id,pt
  - source-table: mysql_test.tbl2
    partition-keys: id,pt

source:
  type: mysql
  name: MySQL Source
  hostname: ${mysql.hostname}
  port: ${mysql.port}
  username: ${mysql.username}
  password: ${mysql.password}
  tables: mysql_test.\.*
  server-id: 8601-8604
  #（可选）同步增量阶段新创建的表的数据
  scan.binlog.newly-added-table.enabled: true
  #（可选）同步表注释和字段注释
  include-comments.enabled: true
  #（可选）优先分发无界的分片以避免可能出现的TaskManager OutOfMemory问题
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #（可选）开启解析过滤，加速读取
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  #（可选）提交用户名，建议为不同作业设置不同的提交用户以避免冲突
  commit.user: your_job_name
  #（可选）开启删除向量，提升读取性能
  table.properties.deletion-vectors.enabled: true
  
transform:
  - source-table: mysql_test.tbl1
    #（可选）设置分区字段
    partition-keys: id,pt
    #（可选）实现软删除
    projection: \*, __data_event_type__ AS op_type
    converter-after-transform: SOFT_DELETE
  - source-table: mysql_test.tbl2
    #（可选）设置分区字段
    partition-keys: id,pt
    #（可选）实现软删除
    projection: \*, __data_event_type__ AS op_type
    converter-after-transform: SOFT_DELETE

OSS 多模态数据实时入湖

Flink CDC支持通过OSS CDC（公测中）实时感知OSS中的文件变更。以图片数据为例，作业可以读取图片内容，调用Flink AI服务（内置模型）生成向量，并将图片元数据、原始内容和向量实时写入DLF Paimon表，为图片检索、内容理解等多模态场景提供统一的数据基础。整体链路为：OSS对象创建事件 → MNS队列 → Flink OSS CDC → FETCH_CONTENT → AI_IMAGE_EMBED → DLF Paimon。

前提条件：

• 实时计算引擎版本为VVR 11.8.0及以上。

• 已创建OSS Bucket、MNS队列和DLF Paimon Catalog，且OSS、MNS和Flink工作空间位于同一地域。

• 已配置OSS对象创建事件并投递到MNS队列；作业运行身份具有OSS读取、MNS消费、DLF写入和Flink AI服务（内置模型）调用权限。

• 通过FETCH_CONTENT函数访问OSS时，还需在作业运行参数中配置Bucket鉴权信息。

以下示例先全量扫描OSS cdc-image-search/images/flowers路径下的存量图片，再通过MNS事件实时感知新增图片，生成图片向量并写入DLF Paimon表：

source:
  type: oss-cdc
  name: OSS CDC Source
  endpoint: http://<yourAccountId>.mns.<yourRegion>-internal.aliyuncs.com
  region: <yourRegion>
  queue-name: cdc-image-search-queue
  access-key-id: ${secret_values.mns_ak_id}
  access-key-secret: ${secret_values.mns_ak_secret}
  scan.startup.mode: INITIAL
  oss-endpoint: https://oss-<yourRegion>-internal.aliyuncs.com
  oss-bucket: <yourBucketName>
  path: cdc-image-search/images/flowers/

transform:
  - source-table: <yourBucketName>
    projection: >-
      key, url, fileName, eTag,
      FETCH_CONTENT(`url`) AS blob_field,
      AI_IMAGE_EMBED(FETCH_CONTENT(`url`), 'qwen3-vl-embedding') AS embedding
    table-options: blob-field=url,blob_field;blob-descriptor-field=url;blob-as-descriptor=true;row-tracking.enabled=true;data-evolution.enabled=true

route:
  - source-table: <yourBucketName>
    sink-table: image_search.image_assets

sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  
pipeline:
  name: <pipeline-name>

  model:
    name: image_embedding_model
    type: openai-compatible
    model: qwen3-vl-embedding

作业启动后，指定OSS路径中的存量图片将首先完成全量入湖，后续上传的新图片将通过OSS事件通知实时写入DLF Paimon表。目标表中包含对象Key、URL、文件名、原始图片内容和Embedding向量。如需进一步构建DLF Global Index并通过VECTOR_SEARCH实现以图搜图，请参见AI全模态入湖实时以图搜图。

Kafka CDC数据实时同步到数据湖DLF

假设Kafka的inventory Topic中存储了两张表（customers和products）的变更数据，且数据格式为Debezium JSON。以下示例作业可将这两张表的数据分别同步到DLF对应的目标表：

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: ${kafka.bootstrap.servers}
  topic: inventory
  scan.startup.mode: earliest-offset
  value.format: debezium-json
  debezium-json.distributed-tables: true

sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  #（可选）提交用户名，建议为不同作业设置不同的提交用户以避免冲突
  commit.user: your_job_name
  #（可选）开启删除向量，提升读取性能
  table.properties.deletion-vectors.enabled: true

# debezium-json不包含主键信息，需要另外为表添加主键  
transform:
  - source-table: \.*.\.*
    projection: \*
    primary-keys: id

Kafka 日志数据实时同步到数据湖DLF

如果您的Kafka集群中存储的是自定义的JSON格式的数据，您可以配置CDC YAML作业同步Kafka数据到DLF存储，我们会为您提供自动数据类型推导、表结构推导和表结构演进的支持。

假设Kafka的inventory Topic中存储了一张日志表的数据，且数据格式为JSON。以下示例作业可将这张表的数据同步到DLF对应的目标表：

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: ${kafka.bootstrap.servers}
  topic: inventory
  scan.startup.mode: earliest-offset
  value.format: json
  # （可选）递归式地展开JSON中的嵌套列
  json.infer-schema.flatten-nested-columns.enable: true
  # （可选）跳过前 100 次出现的解析异常；若超过 100 次则作业失败。
  ingestion.ignore-errors: true
  ingestion.error-tolerance.max-count: 100

sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  #（可选）开启删除向量，提升读取性能
  table.properties.deletion-vectors.enabled: true

# 为表添加主键信息 
transform:
  - source-table: \.*.\.*
    projection: \*
    primary-keys: id
    
# 将 inventory topic 中所有的数据都写入到 test_database.inventory 表中
route:
  - source-table: inventory
    sink-table: test_database.inventory
    
pipeline:
  # （可选）将会导致处理异常的脏数据记录到日志中
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

假设Kafka的inventory Topic中存储了多张日志表的数据，数据格式为JSON，且在JSON内容中的databaseName、tableName字段中提供了库名、表名信息。以下示例作业可将这个Topic中的多张表的数据同步到DLF对应的目标表：

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: ${kafka.bootstrap.servers}
  topic: inventory
  scan.startup.mode: earliest-offset
  value.format: json
  # （可选）递归式地展开JSON中的嵌套列
  json.infer-schema.flatten-nested-columns.enable: true
  # 使用 databaseName 字段中的值作为库名，使用 tableName 字段中的值作为表名
  json.decode.parser-table-id.fields: databaseName,tableName
  # （可选）跳过前 100 次出现的解析异常；若超过 100 次则作业失败。
  ingestion.ignore-errors: true
  ingestion.error-tolerance.max-count: 100

sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  #（可选）开启删除向量，提升读取性能
  table.properties.deletion-vectors.enabled: true

# 为表添加主键信息 
transform:
  - source-table: \.*.\.*
    projection: \*
    primary-keys: id
    
# 将 ods.inventory、ods.customer，ods.user 中的数据分别写入到 test_database.inventory，test_database.customer，test_database.user 表中
route:
  - source-table: ods.inventory
    sink-table: test_database.inventory
  - source-table: ods.customer
    sink-table: test_database.customer
  - source-table: ods.user
    sink-table: test_database.user   
    
pipeline:
  # （可选）将会导致处理异常的脏数据记录到日志中
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

如果您希望了解更多JSON格式的Kafka源表的表结构推导与演进策略，可以参考Kafka SQL连接器说明。