MongoDB数据源

MongoDB Reader支持的MongoDB数据类型

数据集成支持大部分MongoDB类型，但也存在部分不被支持的情况，请注意检查您的数据类型。

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

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

  类型	离线读（MongoDB Reader）	说明
  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属性，进行自定义处理。

  类型	离线读（MongoDB Reader）	说明
  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支持数组类型，并且数组类型具有强大的索引功能。您可以通过参数的特殊配置，将字符串转换为MongoDB中的数组。转换类型后，即可并行写入MongoDB。

数据集成特殊数据类型：combine

MongoDB Reader数据类型转换

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

MongoDB Writer数据类型转换

数据类型示例1：Combine类型使用示例

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结果如下所示。

数据类型示例2：递归解析处理多层嵌套的Document

当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。

单表离线同步任务配置指导

• 操作流程请参见向导模式配置、脚本模式配置。

• 脚本模式配置的全量参数和脚本Demo请参见下文的附录：MongoDB脚本Demo与参数说明。

单表实时同步任务配置指导

操作流程请参见数据集成侧实时同步任务配置、DataStudio侧实时同步任务配置。

整库级别同步任务配置指导

整库离线、整库（实时）全增量、整库（实时）分库分表等整库级别同步任务的配置操作，请参见整库离线同步任务、整库实时同步任务配置。

离线任务脚本配置方式

如果您配置离线任务时使用脚本模式的方式进行配置，您需要按照统一的脚本格式要求，在任务脚本中编写相应的参数，详情请参见脚本模式配置，以下为您介绍脚本模式下数据源的参数配置详情。

Reader脚本Demo

配置一个从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":{
        "common": { 
            "column": { 
                "timeZone": "GMT+0" //时区
            } 
        },
        "errorLimit":{
            "record":"0"//错误记录数。
        },
        "speed":{
            "throttle":true,//当throttle值为false时，mbps参数不生效，表示不限流；当throttle值为true时,表示限流。
            "concurrent":1 //作业并发数。
            "mbps":"12"//限流，此处1mbps = 1MB/s。
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

Writer脚本Demo

配置写入MongoDB的数据同步作业，详情请参见下文的参数说明。

{
    "type": "job",
    "version": "2.0",//版本号。
    "steps": [
        {
            "stepType": "stream",
            "parameter": {},
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "mongodb",//插件名。
            "parameter": {
                "datasource": "",//数据源名。
                "column": [
                    {
                        "name": "_id",//列名。
                        "type": "ObjectId"//数据类型。如果replacekey为_id，则此处的type必须配置为ObjectID。如果配置为string，会无法进行替换。
                    },
                    {
                        "name": "age",
                        "type": "int"
                    },
                    {
                        "name": "id",
                        "type": "long"
                    },
                    {
                        "name": "wealth",
                        "type": "double"
                    },
                    {
                        "name": "hobby",
                        "type": "array",
                        "splitter": " "
                    },
                    {
                        "name": "valid",
                        "type": "boolean"
                    },
                    {
                        "name": "date_of_join",
                        "format": "yyyy-MM-dd HH:mm:ss",
                        "type": "date"
                    }
                ],
                "writeMode": {//写入模式。
                    "isReplace": "true",
                    "replaceKey": "_id"
                },
                "collectionName": "datax_test"//连接名称。
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "setting": {
        "errorLimit": {//错误记录数。
            "record": "0"
        },
        "speed": {
            "throttle": true,//当throttle值为false时，mbps参数不生效，表示不限流；当throttle值为true时,表示限流。
            "concurrent": 1,//作业并发数。
            "mbps": "1"//限流的速度，此处1mbps = 1MB/s。
        }，
       "jvmOption": "-Xms1024m -Xmx1024m"
    },
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    }
}

Writer脚本参数

After applying the update, the (immutable) field '_id' was found to have been altered to _id: "2"

执行数据集成作业时，会首先执行您已配置的preSql。完成preSql的执行后，才可以进入实际的数据写出阶段。preSql本身不会影响写出的数据内容。数据集成通过preSql参数，可以具备幂等执行特性。例如，您的preSql在每次任务执行前都会清理历史数据（根据您的业务规则进行清理）。此时，如果任务失败，您只需要重新执行数据集成作业即可。

preSql的格式要求如下：

• 需要配置type字段，表示前置操作类别，支持drop和remove，例如"preSql":{"type":"remove"}：

  • drop：表示删除集合和集合内的数据，collectionName参数配置的集合即是待删除的集合。

  • remove：表示根据条件删除数据。

  • json：您可以通过JSON控制待删除的数据条件，例如"preSql":{"type":"remove", "json":"{'operationTime':{'$gte':ISODate('${last_day}T00:00:00.424+0800')}}"}。此处的${last_day}为DataWorks调度参数，格式为$[yyyy-mm-dd]。您可以根据需要具体使用其它MongoDB支持的条件操作符号（$gt、$lt、$gte和$lte等）、逻辑操作符（and和or等）或函数（max、min、sum、avg和ISODate等）。

    数据集成通过如下MongoDB标准API执行您的数据，删除query。

    query=(BasicDBObject) com.mongodb.util.JSON.parse(json);        
    col.deleteMany(query);

    说明

    如果您需要条件删除数据，建议优先使用JSON配置形式。

  • item：您可以在item中配置数据过滤的列名（name）、条件（condition）和列值（value）。例如"preSql":{"type":"remove","item":[{"name":"pv","value":"100","condition":"$gt"},{"name":"pid","value":"10"}]}。

    数据集成会基于您配置的item条件项，构造查询query条件，进而通过MongoDB标准API执行删除。例如col.deleteMany(query);。

• 不识别的preSql，无需进行任何前置删除操作。