流式计算 Flink 基于 Flink CDC，通过开发 YAML 作业的方式有效地实现了将数据从源端同步到目标端的数据摄入工作。本文将为您介绍 Flink CDC 作业开发的操作步骤。

前提条件 #

项目管理员（Project_Admin）已经在项目内创建好 Flink 资源池，请参见创建资源池。

体验流程 #

本文将 Flink CDC 任务开发流程总体分为了两个阶段，分别是开发阶段和运维阶段，每个阶段对应的成员及主要工作均不相同。

• 开发阶段：一般是开发人员（Project_Dev）负责开发 CDC 任务，完成开发和调试后将任务上线到生产环境。
• 运维阶段：一般是运维人员（Project_OPS）负责启动任务，并查看任务执行情况。

步骤一：开发 CDC 任务 #

1. 登录流式计算 Flink 版控制台。
2. 在顶部菜单栏选择目标地域。
3. 在左侧导航栏选择项目管理，然后单击项目区块进入项目。
4. 在作业开发页面单击加号按钮，创建任务。
   您也可以选择目标文件夹，直接在该文件夹中创建任务；也可以直接单击 Flink 任务下的 Flink CDC 作业。

5. 在创建作业对话框，设置作业名称、作业类型、存储位置、引擎版本等参数，可以通过预览 CDC 的任务后，确认无误单击立即创建。

6. 在任务编辑区编辑 CDC 任务的业务逻辑代码。
   此处提供一个示例 CDC YAML 任务代码。代码含义为：新建一个 MySQL 类型的数据源，将 app_db 库下所有数据表同步到 doris 数据下游。并且对数据表做一些路由和字段转换工作。

   source:
        type: mysql
        hostname: localhost
        port: 3306
        username: root
        password: 123456
        tables: app_db.\.*
   
      sink:
        type: doris
        fenodes: 127.0.0.1:8030
        username: root
        password: ""
   
      transform:
        - source-table: adb.web_order01
          projection: \*, UPPER(product_name) as product_name
          filter: id > 10 AND order_id > 100
          description: project fields and filter
        - source-table: adb.web_order02
          projection: \*, UPPER(product_name) as product_name
          filter: id > 20 AND order_id > 200
          description: project fields and filter
   
      route:
        - source-table: app_db.orders
          sink-table: ods_db.ods_orders
        - source-table: app_db.shipments
          sink-table: ods_db.ods_shipments
        - source-table: app_db.products
          sink-table: ods_db.ods_products
   
      pipeline:
        name: Sync MySQL Database to Doris
        parallelism: 2
   

7. 参考 使用 JDBC 或者 MySQL-CDC 数据源上传 mysql-connector-java 到依赖文件中

8. CDC 任务代码编辑完成后，单击验证按钮。
   系统会自动校验您的 YAML 语句正确性，如果报错，请根据提示自主完成 YAML 语句修改。检验通过后，系统提示success。
   
9. 代码编辑和验证通过后，单击保存按钮，保存任务代码。

步骤二：上线 CDC 任务 #

开发与生产环境隔离，当任务开发人员完成任务开发和调试后，可以将任务上线到生产环境。

1. 在作业开发栏目下查找并单击目标任务，然后在编辑区上方选择正确的执行方式和引擎版本，再单击上线。

2. 在任务上线设置对话框，选择运行资源池、设置任务优先级和调度策略，然后单击确定。
   系统会提示任务上线成功，可以前往任务管理页面查看。
   

   配置	说明
   运行资源池	从下拉列表中选择任务运行的 Flink 资源池。 注意 如果您提交的任务开启了自动调优，则必须运行在按量付费类型的资源池。
   任务优先级	系统默认预置的优先级为 L3，您可以按需设置任务优先级，数字越小优先级越高。 任务优先级决定了任务内部的调度顺序，优先级高的任务先被调度，即 L3 先于 L4 被调度。
   调度策略	根据需求配置任务调度策略： GANG：保证任务的所有实例被一起调度，即当剩余资源满足任务正常运行所需资源时才进行分配；不满足所需资源则不分配。 该策略不会出现分配资源后，任务却不能启动的现象，解决了资源死锁问题。 DRF：从多维资源考虑，更为合理地将资源公平分配给资源池内的各个任务，从而提升利用率。 例如：剩余10 核 40 GB 的资源，A 任务需要10 核 20 GB 资源；B 任务需要 2 核 8 GB 的资源。如果分配给 A，剩余 0 核 20 GB 资源无法被利用；DRF 策略会选择分配给 B，剩下 8 核 32 GB 可以继续给后来任务使用。
   调度时长	设置为 GANG 调度策略时，需要设置调度时长。调度时长表示再次调度的时间间隔，即任务拉起不成功会再次重试调度。 如果超过调度时长，任务就会调度失败。如果设置为 0，则会一直重试。
   更多配置	CDC 任务支持在上线前跳过深度检查，允许任务强制上线。 默认不勾选。勾选跳过上线前检查后，表示将跳过 CDC 代码深度检查，任务将会直接上线。

步骤四：启动 CDC 任务 #

开发与生产环境隔离，任务开发人员将任务上线到生产环境后，由运维人员启动任务。

1. 在项目左侧导航栏选择任务运维 > 任务管理。

2. 在任务列表页面，单击操作列中的启动。

3. 在启动任务对话框，选择任务启动方式，然后单击确定。
   任务启动需要一定时长，请耐心等待。启动成功后，状态为运行中。
   

   配置	说明
   启动方式	请根据实际情况选择任务启动方式： 从最新状态启动：以最新的 Checkpoint 或 Savepoint 启动（按时间取最新的）。 全新启动：不使用 Checkpoint 或 Savepoint，直接启动。 指定快照启动：指定目标快照（Savepoint）启动。 说明 首次上线的任务，只能是全新启动方式。
   参数配置	任务携带在开发侧的并行度、Task Manager 和 Job Manager 的资源配置。在启动任务时支持您更新配置并快速生效。 说明 更新参数配置并启动任务后，将新增一个任务版本，并将最新配置同步到任务开发侧。 并行度：任务全局并发数。 单个 TaskManager CPU 数：单个 TaskManager 的 CPU 核数。 单个 TaskManager 内存大小：单个 TaskManager 占用的内存大小。 单个 TaskManager slot 数：单个 TaskManager 的 Slot 数量。 JobManager CPU 数：单个 JobManager 的 CPU 核数。 JobManager 内存大小：单个 JobManager 占用的内存大小。
   更多设置	在任务开发变更时新增或修改算子，可能会导致任务无法从快照恢复，此时您可以选择启用允许忽略部分算子状态功能，保证任务能正常运行。 注意 仅当选择指定快照启动或从最新状态启动方式时，支持勾选忽略部分算子状态。 当您选择全新启动方式时，不支持忽略算子状态。

步骤五：查看 Flink UI #

任务在生产环境上正常运行后，您可以在 Flink UI 界面了解任务运行、TaskManager、JobManager 的详细信息。

1. 在项目左侧导航栏选择任务运维 > 任务管理。
2. 在任务列表页面筛选目标任务，单击操作列下的 Flink UI。
   浏览器将会自动打开 Apache Flink Dashboard 页面。
3. 在 Apache Flink Dashboard 左侧导航栏选择 Task Managers，然后单击任务 ID。
   
4. 在任务的 Task Manager 详情页面，单击 Stdout 页签，观察作业 Task 正常，并可以通过控制台查看任务执行输出以及日志等信息确认业务正常。