在进行自建Flink作业迁移前，需要先搭建基础环境。基础环境用于演示作业迁移的过程，如果您已经有了自建Flink环境，则可以不用搭建基础环境。本文为您介绍如何搭建基础环境。

背景信息

CADT是一个云资源创建工具，借助该工具，您可以快速开通各种云资源，搭建好迁移环境。当然，您也可以在各云产品控制台开通云资源，和在CADT上效果一样。本文将按以下步骤为您介绍如何搭建基础环境并测试环境：

步骤一：使用云速搭CADT创建资源

  1. 登录CADT控制台，创建应用。
    1. 登录CADT控制台。
    2. 在顶部菜单栏，选择新建 > 官方模板库新建。
      模板
    3. 在搜索框中，输入Flink迁移后，按回车键。
    4. 鼠标悬浮在开源Flink迁移实时计算Flink全托管版上，单击基于应用新建。
  2. 修改EMR-Flink集群密码。
    说明 本文使用EMR-Flink作为用户自建Flink集群来运行迁移前的作业，以此来对比验证自建Flink迁移后的作业运行结果。如果您已经有了自建Flink集群，可以忽略此步骤。
    1. 鼠标左键双击EMR-Flink集群图标。
    2. 在右侧面板上，修改EMR-Flink集群密码。
      设置EMR密码
    3. 在页面右上角，单击保存。
    4. 输入应用名称。
    5. 单击确认。
  3. 修改Kafka实例版本值。
    1. 鼠标左键双击Kafka实例图标。
    2. 修改部署实例-版本信息。
      kafka版本
    3. 单击保存。
  4. 部署应用。
    1. 在页面右上角，单击部署应用。
    2. 单击下一步：价格清单。
    3. 单击下一步：确认订单。
    4. 选中云速搭服务条款。
    5. 单击下一步：支付并创建。
      下一步
      创建成功后应用如下图。部署成功

步骤二：在RDS中，创建数据库账号和测试数据库

  1. 登录RDS控制台。
    1. 鼠标左键双击RDS实例图标。
      RDS图标
    2. 单击返回。
      返回
    3. 单击资源名称。
      RDS
  2. 创建数据库账号。
    1. 在左侧导航栏，单击账号管理。
    2. 单击创建账号。
    3. 填写账号和密码信息。
      创建账号和密码
    4. 填写完成后，单击确定。
  3. 创建测试数据库test_db。
    1. 在左侧导航栏，单击数据库管理。
    2. 单击创建数据库。
    3. 填写数据库信息。
      数据库
    4. 单击创建。

步骤三：在Kafka中，准备测试数据

这里将通过Flink全托管往Kafka实时写入数据的方式准备测试数据。

  1. 登录Kafka控制台。
    1. 鼠标左键双击Kafka实例图标。
      Kafka
    2. 单击返回。
      返回
    3. 单击资源名称。
      kafka
  2. 创建一个名称为kafka-order的Kafka Topic。
    1. 在左侧导航栏，单击Topic管理。
    2. 单击创建Topic。
    3. 填写Topic信息。
      topic
    4. 单击确定。
  3. 将Flink全托管所在VPC的网段添加到Kafka白名单中。
    1. 在左侧导航栏，单击实例详情。
    2. 在接入点信息区域，单击编辑白名单。
    3. 将Flink全托管所在VPC的网段信息填入白名单。
    4. 单击确定。

步骤四：创建Flink SQL作业并上线后测试数据生产

这里使用Flink往Kafka实时写入测试数据，首先创建Flink作业。说明：源表为Kafka中的订单表，topic名为kafka-order，存储格式为 CSV，Schema 如下。Schema
  1. 登录Flink全托管控制台。
    1. 鼠标左键双击Flink全托管图标。
      Flink全托管
    2. 单击返回。
      返回
    3. 单击资源名称。
      Flink-on-cloud
    4. 单击控制台。
  2. 创建Flink SQL流作业。
    1. 在左侧导航栏，单击作业开发。
    2. 单击新建。
    3. 在新建文件对话框，填写作业配置信息。
    4. 在作业开发页面，编写DDL和DML代码。
      代码示例如下。
      CREATE TEMPORARY TABLE data_in (
  id VARCHAR, 
  order_value FLOAT
) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '100',
  'fields.id.length' = '10',
  'fields.order_value.min' = '1.0',
  'fields.order_value.max' = '100.0'
);

CREATE TEMPORARY TABLE kafka_order (
  order_id VARCHAR,
  order_time TIMESTAMP,
  order_type VARCHAR,
  order_value FLOAT
) WITH (
  'connector' = 'kafka',
  'topic' = 'kafka-order',
  'properties.bootstrap.servers' = '192.*.*.224:9092,192.*.*.225:9092,192.*.*.226:9092',
  'format' = 'csv'
);

INSERT INTO kafka_order
SELECT id as order_id,
  CURRENT_TIMESTAMP as order_time,
  CASE
    WHEN (substring (id, 1, 1) <= 'z' AND substring (id, 1, 1) >= 'a') and (substring (id, 2, 1) <= 'z' AND substring (id, 2, 1) >= 'a') THEN 'typeA'
    WHEN (substring (id, 1, 1) <= 'z' AND substring (id, 1, 1) >= 'a') and (substring (id, 2, 1) <= '9' AND substring (id, 2, 1) >= '0') THEN 'typeB'
    WHEN (substring (id, 1, 1) <= '9' AND substring (id, 1, 1) >= '0') and (substring (id, 2, 1) <= '9' AND substring (id, 2, 1) >= '0') THEN 'typeC'
    ELSE 'typeD'
  END as order_type,
  order_value
FROM
  data_in;
  3. 修改Kafka参数properties.bootstrap.servers的取值。
    请填写为Kafka实例详情的接入点信息中查看到的Kafka接入点信息。
  4. 单击上线。
  5. 在作业运维页面，单击启动。

步骤五：在Kafka控制台查看测试数据

  1. 登录Kafka控制台。
    1. 鼠标左键双击Kafka实例图标。
      Kafka
    2. 单击返回。
      返回
    3. 单击资源名称。
      kafka
  2. 在Kafka Topic中，查看Flink全托管写入的消息。
    1. 在左侧导航栏，单击Topic管理。
    2. 单击kafka_order。
    3. 在消息查询页签，查询消息。
      查询消息
      说明 等Flink作业启动运行后，在Kafka Topic中就可以看到测试数据已生成。