大数据开发治理平台 DataWorks：MongoDB Reader

对于支持读取的数据类型，数据集成在读取时：

• 基本类型的数据，会根据同步任务配置的读取字段（column，详见下文的全量参数说明）中的name自动读取对应path下的数据，并根据数据类型做自动转换，您无需指定column的type属性。

  类型	是否支持	说明
  ObjectId	支持	对象ID类型。
  Double	支持	64位浮点数类型。
  32-bit integer	支持	32位整数。
  64-bit integer	支持	64位整数。
  Decimal128	支持	Decimal128类型。 说明 如果配置为嵌套类型、Combine类型，JSON序列化时会被当做对象处理，需增加参数decimal128OutputType为bigDecimal，才能输出为decimal。
  String	支持	字符串类型。
  Boolean	支持	布尔类型。
  Timestamp	支持	时间戳类型。 说明 BsonTimestamp存储的是时间戳，无需考虑时区影响，详情请参见MongoDB中的时区问题。
  Date	支持	日期类型。

• 部分复杂类型的数据，您可通过配置column的type属性，进行自定义处理。

  类型	是否支持	说明
  Document	支持	嵌入文档类型。 如果没有配置type属性，则直接将Document转JSON序列化处理。 如果配置了type属性为document，则属于嵌套类型，MongoDB Reader会按path读取Document属性。详细示例请参见下文的数据类型示例2：递归解析处理多层嵌套的Document。
  Array	支持	数组类型。 如果type配置为array.json、arrays，直接JSON序列化处理。 如果type配置为array、document.array，则拼接为字符串，分隔符（column中的splitter）默认为英文逗号。

结合上文可见，MongoDB Reader针对MongoDB类型的转换列表，如下表所示。

转换后的类型分类	MongoDB数据类型
LONG	INT、LONG、document.INT和document.LONG
DOUBLE	DOUBLE和document.DOUBLE
STRING	STRING、ARRAY、document.STRING、document.ARRAY和COMBINE
DATE	DATE和document.DATE
BOOLEAN	BOOL和document.BOOL
BYTES	BYTES和document.BYTES

MongoDB Reader插件的Combine数据类型支持将MongoDB document中的多个字段合并成一个JSON串。例如，导入MongoDB中的字段至MaxCompute，有字段如下（下文均省略了value使用key来代替整个字段）的三个document，其中a、b是所有document均有的公共字段，x_n是不固定字段。

• doc1： a b x_1 x_2
• doc2: a b x_2 x_3 x_4
• doc3: a b x_5

配置文件中要明确指出需要一一对应的字段，需要合并的字段则需另取名称（不可以与document中已存在字段同名），并指定类型为COMBINE，如下所示。

"column": [
{
"name": "a",
"type": "string",
},
{
"name": "b",
"type": "string",
},
{
"name": "doc",
"type": "combine",
}
]

最终导出的MaxCompute结果如下所示。

odps_column1	odps_column2	odps_column3
a	b	{x_1,x_2}
a	b	{x_2,x_3,x_4}
a	b	{x_5}

当MongoDB中Document存在多层嵌套时，可通过配置document类型进行递归解析处理。示例如下：

• MongoDB源端数据为：

  {
      "name": "name1",
      "a":
      {
          "b":
          {
              "c": "this is value"
          }
      }
  }

• MongoDB列可配置为：

  {"name":"_id","type":"string"}
  {"name":"name","type":"string"}
  {"name":"a.b.c","type":"document"}

  

如上配置，可将源端嵌套字段a.b.c的值写入目标端c字段中，同步任务运行后，目标端写入数据为this is value。

您需要在数据同步任务的编辑页面进行以下配置：

1. 选择数据源。
   配置同步任务的数据来源和数据去向。

   参数	描述
   数据源	即上述参数说明中的datasource，通常输入您配置的数据源名称。
   集合名称	即上述参数说明中的collectionName。
   批量条数	从MongoDB批量获取的数据条数，默认值为1000。
   游标超时时间	游标超时时间，默认值是3600000毫秒，如果配置为负数，则永不超时。
   检索查询条件	即上述参数说明中的query。您可以使用调度参数来指定同步源表的数据范围，当同步任务运行时，任务中配置的参数都会被替换为调度参数表达式所表达的实际值，然后再执行数据同步。离线同步增量同步实现方式请参见数据集成使用调度参数的相关说明。 说明 非时间类型增量字段同步：可以通过赋值节点将字段处理为目标数据类型后，再传入数据集成进行数据同步。例如，当MongoDB存储的增量字段为时间戳，您可以通过赋值节点将时间类型字段通过引擎函数转换为时间戳，再传给离线同步任务使用，关于赋值节点的使用详情请参见：赋值节点。

2. 字段映射，即上述参数说明中的column。默认使用同行映射。您可以单击图标手动编辑源表字段。
   
3. 通道控制。

   参数	描述
   任务期望最大并发数	数据同步任务内，可以从源并行读取或并行写入数据存储端的最大线程数。向导模式通过界面化配置并发数，指定任务所使用的并行度。 说明 当前任务期望最大并发数仅支持配置为1。
   同步速率	设置同步速率可以保护读取端数据库，以避免抽取速度过大，给源库造成太大的压力。同步速率建议限流，结合源库的配置，请合理配置抽取速率。
   错误记录数	错误记录数，表示脏数据的最大容忍条数。
   分布式处理能力	数据同步时，可以将任务切片分散到多台执行节点上并发执行，提高同步速率。该模式下，配置较大任务并发数会增加数据存储访问压力，如需使用该功能，请提前评估数据存储的访问负载。该功能仅支持在独享数据集成资源组配置，详情请参见独享数据集成资源组和新增和使用独享数据集成资源组。

配置一个从MongoDB抽取数据到本地的作业，详情请参见上述参数说明。

{
    "type":"job",
    "version":"2.0",//版本号。
    "steps":[
        {
            "category": "reader",
            "name": "Reader",
            "parameter": {
                "datasource": "datasourceName", //数据源名称。
                "collectionName": "tag_data", //集合名称。
                "query": "", // 数据查询过滤。
                "column": [
                    {
                        "name": "unique_id", //字段名称。
                        "type": "string" //字段类型。
                    },
                    {
                        "name": "sid",
                        "type": "string"
                    },
                    {
                        "name": "user_id",
                        "type": "string"
                    },
                    {
                        "name": "auction_id",
                        "type": "string"
                    },
                    {
                        "name": "content_type",
                        "type": "string"
                    },
                    {
                        "name": "pool_type",
                        "type": "string"
                    },
                    {
                        "name": "frontcat_id",
                        "type": "array",
                        "splitter": ""
                    },
                    {
                        "name": "categoryid",
                        "type": "array",
                        "splitter": ""
                    },
                    {
                        "name": "gmt_create",
                        "type": "string"
                    },
                    {
                        "name": "taglist",
                        "type": "array",
                        "splitter": " "
                    },
                    {
                        "name": "property",
                        "type": "string"
                    },
                    {
                        "name": "scorea",
                        "type": "int"
                    },
                    {
                        "name": "scoreb",
                        "type": "int"
                    },
                    {
                        "name": "scorec",
                        "type": "int"
                    },
                    {
                        "name": "a.b",
                        "type": "document.int"
                    },
                    {
                        "name": "a.b.c",
                        "type": "document.array",
                        "splitter": " "
                    }
                ]
            },
            "stepType": "mongodb"
        },
        { 
            "stepType":"stream",
            "parameter":{},
            "name":"Writer",
            "category":"writer"
        }
    ],
    "setting":{
        "errorLimit":{
            "record":"0"//错误记录数。
        },
        "speed":{
            "throttle":true,//当throttle值为false时，mbps参数不生效，表示不限流；当throttle值为true时,表示限流。
            "concurrent":1 //作业并发数。
            "mbps":"12"//限流
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

在数据读取时，用户配置的Column.name为大小写敏感，如配置有误，会导致读出数据为null。例如：

• MongoDB源数据为：

  {
      "MY_NAME": "zhangsan"
  }

• 同步任务的Column配置为：

  {
      "column":
      [
          {
              "name": "my_name"
          }
      ]
  }

则由于同步任务的配置与源数据大小写不一致，会导致数据读取异常。