本文介绍通过应用集成实现从MongoDB获取数据写入Apache Kudu的数据集成转换。完成从MongoDB指定集合获取数据,然后写入Apache Kudu目标表,过程中如果两者数据结构存在差异可以加入数据映射逻辑步骤。

前提条件

  • 在阿里云容器服务Kubernetes版上创建了MongoDB和Apache Kudu服务,详细操作步骤请参见使用镜像快速创建无状态Deployment应用
  • 已经在MongoDB上创建game集合,并在game集合中存入数据。
  • 已经在Apache Kudu上创建test表。

背景信息

通过集成流配置可以实现从MongonDB指定集合中获取数据,并将数据写入到Apache Kudu的指定表中。

通过创建的集成,将完成以下动作:
  1. 从MongonDB的game集合获取数据。
  2. 将获取到的数据写入到Apache Kudu的test表中。

创建连接

本示例中会用到MongoDB和Apache Kudu,所以需要借助连接器创建对应的连接。

  • 创建MongoDB连接,详情请参见创建连接

    MongoDB连接器的参数说明请参见MongoDB

  • 创建Apache Kudu连接,详情请参见创建连接

创建空白集成

  1. 登录应用集成控制台
  2. 在顶部菜单栏选择地域
  3. 在左侧导航栏选择集成 > 集成列表
  4. 集成列表页面单击新建集成
  5. 新建集成面板中创建方式选择为空白流,选择目标环境,输入名称,然后单击创建
  6. 集成创建后,返回集成设计页面,选择接下来的操作。
    • 在页面左上角选择工作空间,然后在右上角单击保存,创建一个空集成。
    • 在页面左上角单击 ,在列表中单击Flow,创建集成流。详情请参见创建集成流

创建集成流

  1. 集成设计页面设置集成名,选择目标工作空间环境
  2. 集成设计页面左上角单击 ,在列表中单击Flow,创建集成流。
    也可以在页面中单击点击创建,在列表中单击Flow,创建集成流。
  3. 选择触发器,实现从MongoDB的game集合中获取数据。
    1. 创建新集成流对话框输入名称,并选择之前创建的MongoDB作为触发器,然后单击创建
      MongoDB触发器
    2. 选择操作对话框Mongo consumer右侧,单击选择
      选择operation-Mongo consumer
    3. 步骤配置对话框设置Mongo DB消费者参数,然后单击确定
      步骤配置-从MaogoDB读取数据
      Mongo DB消费者参数说明。
      参数 描述
      集合名称 获取数据的集合名称,本场景设置为game
      DB用于存储尾部跟踪 用于存储该跟踪过程的偏移量。
      唯一的id来识别这个跟踪过程 如果在集成中有多个跟踪过程,请指定一个不同的id。
      用于存储尾部跟踪的字段 属性,用于存储此尾部跟踪过程的偏移量。
      用于存储尾部跟踪的集合 字段,用于存储此尾部跟踪过程的偏移量。
      持续跟踪是否启用 是否持续跟踪。
      用于跟踪传入文档的集合字段 用于跟踪新文档的属性,通常是增量id或时间戳,本场景设置为name
    4. 设置outputDataShape对话框中设置参数,然后单击创建
      设置outputDataShape
      设置outputDataShape参数说明。
      参数 描述
      选择scheme 选择Json Schema
      scheme 如果需要,可以输入JSON脚本。
      名称 scheme名称。
      描述 scheme的描述。
      本场景的示例JSON脚本如下。
      {
          "type":"array",
          "$schema":"http://json-schema.org/schema#",
          "items":{
              "type":"object",
              "properties":{
                  "code":{
                      "type":"string",
                      "required":true
                  },
                  "name":{
                      "type":"string",
                      "required":true
                  },
                  "year":{
                      "type":"string",
                      "required":true
                  },
                  "language":{
                      "type":"string",
                      "required":false
                  },
                  "genre":{
                      "type":"string",
                      "required":true
                  }
              }
          }
      }
    创建完成后,集成流中即出现了从MongoDB指定集合获取信息的触发器。Mongo consumer-触发器
  4. 在集成流中添加Apache Kudu连接,将获取到的数据插入到Apache Kudu的test表中。
    1. 在集成流中Mongo consumer后单击
    2. 选择组件类型对话框单击连接,然后单击之前新建的Apache Kudu连接。
    3. 选择操作对话框Insert a row in a kudu table右侧,单击选择
       选择operation-Insert a row in a kudu table
    4. 步骤配置对话框设置Apache Kudu的指定表,然后单击确定
      步骤配置-指定表
    5. 设置inputDataShape对话框选择scheme列表中选择任意类型,然后单击创建
    创建完成后,集成流中即出现了数据上传到Apache Kudu指定表的连接。Mongo consumer-Insert a row in a kudu table
  5. 在集成流中添加DataMapper,完成字段映射。
    1. 在集成流中Mongo consumer后单击
    2. 选择组件类型对话框单击逻辑步骤,然后单击Data Mapper
    3. Source > 1-Mongo > Document RootTarget > 2-Kudu table之间单击字段进行映射,然后单击确认
      字段映射
      字段映关系如下。
      Source字段 Target字段
      name name
      year year
      code grade

      本场景以code字段和grade字段映射为例,您可根据实际需要选择字段映射关系。

      genre
      language
    4. 设置对话框选择scheme列表中选择任意类型,然后单击创建
  6. 集成设计页面右上角单击保存,保存集成流。
    MongonDB处理的集成流创建完成。MongoDB处理
    说明 返回集成设计页面时,请及时在页面右上角单击保存,以免添加的步骤丢失。

部署集成

集成创建并保存后,需要对集成进行部署。详情请参见部署集成示例

结果验证

  • 查看MongoDB集合数据。
    • 登录Apache Kudu,在test表中查看是否含有MongoDB的game集合的数据。
    • 您也可以在MongoDB的game集合新增一条数据,然后查看Apache Kudu的test表是否有新增的数据。
  • 验证集成部署是否正常运行。

    查看目标集成的执行记录,其执行日志状态是否为SUCCESS