DataWorks:MongoDB data source

MongoDB data types supported by MongoDB Reader

Data Integration supports most MongoDB data types. However, some data types are not supported. Verify your data types before proceeding.

For supported data types, Data Integration reads data as follows:

• For primitive data types, Data Integration automatically reads data from the corresponding path based on the name of the read field (column) configured for the sync task. It also automatically converts the data type. You do not need to specify the type property for the column. For more information, see Appendix: MongoDB script demos and parameter descriptions.

  Type	Offline read (MongoDB Reader)	Description
  ObjectId	Supported	The object ID type.
  Double	Support	The 64-bit floating-point number type.
  32-bit integer	Support	A 32-bit integer.
  64-bit integer	Support	A 64-bit integer.
  Decimal128	Support	The Decimal128 type. Note If this type is configured as a nested type or a Combine type, it is processed as an object during JSON serialization. You must add the decimal128OutputType parameter and set it to bigDecimal to output the data as a decimal.
  String	Supported	The string type.
  Boolean	Support	The Boolean type.
  Timestamp	Support	The timestamp type. Note BsonTimestamp stores a timestamp. You do not need to consider the impact of time zones. For more information, see Time Zone Issues in MongoDB.
  Date	Support	The date type.

• For some complex data types, you can customize processing by configuring the type property of the column.

  Type	Offline read (MongoDB Reader)	Description
  Document	Support	The embedded document type. If the type property is not configured, the document is directly processed by JSON serialization. If the type property is set to document, it is a nested type. MongoDB Reader reads the document properties by path. For a detailed example, see Data type example 2: Recursively parsing multi-level nested documents.
  Array	Support	The array type. If type is set to array.json or arrays, the data is directly processed by JSON serialization. If type is set to array or document.array, the data is concatenated into a string. The default separator (splitter in the column) is a comma (,).

  Important

  Data Integration does not support the array type. However, MongoDB supports arrays and their powerful indexing features. You can configure specific parameters to convert strings into arrays in MongoDB. After conversion, you can write data to MongoDB in parallel.

Special Data Integration data type: combine

MongoDB Reader data type conversions

The following table lists the data type conversions performed by MongoDB Reader.

MongoDB Writer data type conversions

Data type example 1: Using the combine type

The Combine data type of the MongoDB Reader plugin lets you merge multiple fields in a MongoDB document into a single JSON string. For example, you want to import fields from MongoDB to MaxCompute. The following three documents contain fields where the key represents the entire field and the value is omitted. The fields `a` and `b` are common to all documents, and `x_n` are non-fixed fields.

• doc1: a b x_1 x_2

• doc2: a b x_2 x_3 x_4

• doc3: a b x_5

In the configuration file, specify the fields that require one-to-one mapping. For fields to be merged, assign a new name that differs from any existing field in the document, and set the type to COMBINE, as shown below.

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

The final exported result in MaxCompute is as follows.

Data type example 2: Recursively parsing multi-level nested documents

When a MongoDB document has multiple levels of nesting, you can configure the document type to process it recursively. The following is an example:

• Source data in MongoDB:

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

• MongoDB column configuration:

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

  

With this configuration, the value of the nested field `a.b.c` from the source is written to the destination field `c`. After the sync task runs, the data written to the destination is this is value.

Configure a single-table offline sync task

• For the procedure, see Configure a task in the codeless UI and Configure a task in the code editor.

• For all parameters and a script demo for configuring a task in the code editor, see Appendix: MongoDB script demos and parameter descriptions.

Configure a single-table real-time sync task

For the procedure, see Configure a real-time sync task in Data Integration and Configure a real-time sync task in DataStudio.

Configure a whole-database sync task

For information about how to configure whole-database sync tasks, such as whole-database offline sync, whole-database full and incremental real-time sync, and whole-database real-time sync with sharding, see Whole-database offline sync tasks and Configure a whole-database real-time sync task.

Configure a batch synchronization task by using the code editor

If you want to configure a batch synchronization task by using the code editor, you must configure the related parameters in the script based on the unified script format requirements. For more information, see Configure a task in the code editor. The following information describes the parameters that you must configure for data sources when you configure a batch synchronization task by using the code editor.

Reader script demo

The following code shows how to configure a job to extract data from MongoDB to a local destination. For more information about the parameters, see the following table.

{
    "type":"job",
    "version":"2.0",// The version number.
    "steps":[
        {
            "category": "reader",
            "name": "Reader",
            "parameter": {
                "datasource": "datasourceName", // The name of the data source.
                "collectionName": "tag_data", // The name of the collection.
                "query": "", // The data query filter.
                "column": [
                    {
                        "name": "unique_id", // The name of the field.
                        "type": "string" // The type of the field.
                    },
                    {
                        "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" // The time zone.
            } 
        },
        "errorLimit":{
            "record":"0"// The number of error records.
        },
        "speed":{
            "throttle":true,// If throttle is set to false, the mbps parameter does not take effect, which indicates that the data rate is not limited. If throttle is set to true, the data rate is limited.
            "concurrent":1, // The number of concurrent jobs.
            "mbps":"12"// The data rate limit. 1 mbps is equal to 1 MB/s.
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

Writer script demo

The following code shows how to configure a data sync job to write data to MongoDB. For more information about the parameters, see the following table.

{
    "type": "job",
    "version": "2.0",// The version number.
    "steps": [
        {
            "stepType": "stream",
            "parameter": {},
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "mongodb",// The name of the plug-in.
            "parameter": {
                "datasource": "",// The name of the data source.
                "column": [
                    {
                        "name": "_id",// The name of the column.
                        "type": "ObjectId"// The data type. If replaceKey is set to _id, the type must be set to ObjectId. If you set the type to string, the replacement fails.
                    },
                    {
                        "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": {// The write mode.
                    "isReplace": "true",
                    "replaceKey": "_id"
                },
                "collectionName": "datax_test"// The name of the connection.
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "setting": {
        "errorLimit": {// The number of error records.
            "record": "0"
        },
        "speed": {
            "throttle": true,// If throttle is set to false, the mbps parameter does not take effect, which indicates that the data rate is not limited. If throttle is set to true, the data rate is limited.
            "concurrent": 1,// The number of concurrent jobs.
            "mbps": "1"// The data rate limit. 1 mbps is equal to 1 MB/s.
        },
       "jvmOption": "-Xms1024m -Xmx1024m"
    },
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    }
}

Writer script parameters

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

When a Data Integration job runs, the configured preSql executes first. The actual data writing phase begins only after preSql completes. The preSql itself does not affect the content of the data being written. The preSql parameter enables Data Integration to support idempotent execution. For example, your preSql can clean up historical data before each task execution according to your business rules. In this case, if the task fails, you only need to re-run the Data Integration job.

The format requirements for preSql are as follows:

• You must configure the type field to indicate the category of the pre-operation. Supported values are drop and remove. For example, "preSql":{"type":"remove"}.

  • drop: Deletes the collection and its data. The collection specified by the collectionName parameter is the one to be deleted.

  • remove: Deletes data based on a condition.

  • json: You can use JSON to define conditions for data deletion. For example, "preSql":{"type":"remove", "json":"{'operationTime':{'$gte':ISODate('${last_day}T00:00:00.424+0800')}}"}. Here, ${last_day} is a DataWorks scheduling parameter in the format $[yyyy-mm-dd]. You can use other MongoDB-supported conditional operators (such as $gt, $lt, $gte, and $lte), logical operators (such as and and or), or functions (such as max, min, sum, avg, and ISODate) as needed.

    Data Integration executes your data operation to delete the query using the following standard MongoDB API.

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

    Note

    To delete data conditionally, we recommend using the JSON configuration format.

  • item: You can configure the column name (name), condition (condition), and column value (value) for data filtering in the item. For example, "preSql":{"type":"remove","item":[{"name":"pv","value":"100","condition":"$gt"},{"name":"pid","value":"10"}]}.

    Data Integration constructs a query condition based on the item conditions you configure and then executes the deletion using the standard MongoDB API. For example, col.deleteMany(query);.

• If the preSql is not recognized, no pre-deletion operation is performed.