Mongo 任务配置

Mongo 是一个基于分布式文件存储的数据库，可以存储比较复杂的数据类型，数据集成离线同步任务支持您将数据从 Mongo 中读取，并写入到其他数据源中。
本文将为您介绍 Mongo 数据库通过离线任务的 DSL（脚本）模式，实现与不同数据源的数据互通能力。

1 使用前提

1. 已完成 Mongo 数据源配置。详见配置 Mongo 数据源。

2. 新建任务需拥有项目下管理员或开发等项目角色。详见管理成员。

2 新建离线任务节点

1. 登录 DataLeap 租户控制台。

2. 在左侧导航栏，单击项目管理，进入项目列表界面。

3. 单击相应的项目名称，进入到数据开发界面。

   

4. 在数据开发界面，单击目录树上新建任务按钮，进入新建任务界面。

5. 选择任务类型：

   1. 分类：数据集成。

   2. 选择任务：离线集成。

6. 填写任务基本信息：

   1. 任务名称：输入任务的名称，只允许字符.、字母、数字、下划线、连字符、[]、【】、（）、()以及中文字符，且需要在127个字符以内。

   2. 保存至： 选择任务存放的目标文件夹目录。

7. 单击确定按钮，完成任务创建。

3 配置 DSL 任务

Mongo 同步任务支持 DSL 模式进行配置，在任务编辑界面，单击上方工具栏切换至脚本模式按钮，进入编辑界面。

参考以下 Reader 脚本，进行 Mongo 任务脚本配置。

• Mongo Reader

  {
      "version": "0.2",//版本号
      "type": "batch",
      "reader": {
          "type": "mongo",//reader端数据源类型
          "datasource_id": xxx,//数据源ID
          "parameter": {
              "collection_name": "collection1",//mongo数据库中的集合名称
              "split_pk": "id",//分片字段，根据split_pk指定的字段进行数据分片，同步时启动并发任务进行数据同步。
              "columns": [//字段名称和类型信息
                  {
                      "type": "int",
                      "name": "id"
                  },
                  {
                      "type": "objectID",
                      "name": "objectID"   
                  },  
                  {
                      "type": "Decimal",
                      "name": "NumberDecimal"
                  },
                  {
                      "type": "array",
                      "name": "array"
                  },
                  {
                      "type": "timestamp",
                      "name": "timestamp"
                  },
                  {
                      "type": "date",
                      "name": "ISOdate"   
                  }     
              ]
          }
      },
      "writer": {
          "type": "hive_emr",//目标端写入的数据源类型
          "datasource_id": xxx,//数据源ID
          "parameter": {
              "ak": "xxxxx",//拥有 hive 读写权限的 ak 信息
              "sk": "xxxxx",//拥有 hive 读写权限的 sk 信息
              "partition":"data=${date}",//写入目的表分区
              "table_name":"table",//表名
              "columns": [//字段名称和类型信息
                 {
                      "type": "bigint",
                      "name": "id"
                  },
                  {
                      "type": "string",
                      "name": "address"
                  },
                  {
                      "type": "double",
                      "name": "price"
                  },
                  {
                      "type": "array < string >",
                      "name": "list_info"
                  },
                  {
                      "type": "timestamp",
                      "name": "timestamp"
                  },
                  {
                      "type": "date",
                      "name": "date_info"
                  }
              ],
              "storage_type": "tos"//hive 中依赖的存储介质为 tos 
          }
      },
      "common": {
          "max_parallelism":1,//期望最大并发数
          "dirty_records_count_threshold":N,//脏数据条数设置，脏数据量超过 N 条时，任务停止运行
          "parameter": {
              "optional": {// 自定义高级参数配置
                  // "key" : "value" (must be string)
              }
          }
      }
  }
  

• 参数介绍

  Reader DSL 模式配置参数说明如下：
    其中参数名称前带 * 的为必填参数，名称前未带 * 的为可选填参数。

  参数	说明
  * datasource_id	数据源ID，可从项目控制台 > 数据源管理界面中查看。
  * type	数据源类型： Mongo Reader：填写 mongo； Hive Writer：填写 hive_emr。
  * collection_name	mongo 数据库中的集合名称。
  * split_pk	从 mongo 读取数据时，指定 split_pk 切分建，根据指定的字段进行数据分片，同步时启动并发任务进行数据同步，以提高数据同步的效能，split_pk 字段推荐使用“主键”。
  * table_name	数据表名称信息。
  * columns	填写表的字段名和字段类型信息。
  storage_type	指定 hive 底层存储为 TOS；不指定时，默认为 HDFS。
  * ak	拥有 hive 读写权限的 ak 信息。
  * sk	拥有 hive 读写权限的 sk 信息。
  * partition	目的表分区字段配置，可以配置表达式形式：data=${date}。更多表达式参见变量说明。
  max_parallelism	期望最大并发数，按照实际独享数据集成资源组规格进行设置。
  dirty_records_count_threshold	脏数据条数设置，脏数据量超过 N 条时，任务停止运行。
  optional	任务运行自定义高级参数填写，以 "key" : "value" 形式输入。

4 导航栏功能区介绍

5 配置调度属性

DSL 脚本配置完成后，您可进行数据集成同步任务的调度设置，使任务在生产环境中，能周期性的调度执行。详见调度设置。

6 提交任务

任务所需参数配置和调试任务成功后，将任务提交发布到运维中心离线任务运维中执行。 单击操作栏中的保存和提交上线按钮，在弹窗中，需先通过提交事前检查和提交上线等上线流程，最后单击确认按钮，完成作业提交。详见概述---离线任务提交发布。

后续任务运维操作详见：离线任务运维。