DataWorks:MongoDB data source

MongoDB data types supported by MongoDB Reader

Data Integration supports most, but not all, MongoDB data types. Make sure that your data types are supported.

When Data Integration reads supported data types, it performs the following operations:

• For primitive data types, Data Integration automatically reads data from the corresponding path based on the name of the field that is configured in the column parameter. For more information, see Appendix: Sample script and parameter description for MongoDB. Data Integration also automatically converts the data type. You do not need to specify the type property for the column.

  Type	Batch read (MongoDB Reader)	Description
  ObjectId	Supported	The object ID type.
  Double	Supported	The 64-bit floating-point number type.
  32-bit integer	Supported	A 32-bit integer.
  64-bit integer	Supported	A 64-bit integer.
  Decimal128	Supported	The Decimal128 type. Note If a field 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	Supported	The Boolean type.
  Timestamp	Supported	The timestamp type. Note BsonTimestamp stores timestamps. 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 configure the type property for the column to perform custom processing.

  Type	Batch read (MongoDB Reader)	Description
  Document	Supported	The embedded document type. If the type property is not configured, the Document is directly converted using JSON serialization. If the type property is set to document, the field is a nested type. MongoDB Reader reads the Document properties based on the path. For a detailed example, see Example 2: Recursively parsing a multi-level nested Document below.
  Array	Supported	The array type. If type is set to array.json or arrays, the data is directly processed using JSON serialization. If type is set to array or document.array, the elements are concatenated into a string. The separator, which is specified in the splitter property of the column, is a comma (,) by default.

  Important

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

Special Data Integration data type: combine

MongoDB Reader data type mappings

The following table lists the mappings between MongoDB data types and Data Integration data types for MongoDB Reader.

MongoDB Writer data type mappings

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, assume that you want to import fields from three MongoDB documents to MaxCompute. In the following example, the fields are represented by keys instead of key-value pairs. The fields a and b are common to all three documents, and x_n is a variable field.

• 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, you must explicitly specify the fields that require one-to-one mapping. For the fields that you want to merge, assign a new name that is different from any existing field name in the document and set the type to COMBINE. The following code provides an example.

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

The following table shows the final output in MaxCompute.

Example 2: Recursively parsing a multi-level nested Document

If a document in MongoDB has multiple levels of nesting, you can configure the document type to recursively parse it. The following code provides 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 the preceding configuration, the value of the nested source field a.b.c is written to the destination field c. After the synchronization task runs, the data that is written to the destination is this is value.

Configure a batch synchronization task for a single table

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

• For more information about all parameters and a sample script for the code editor, see Appendix: Sample script and parameter description for MongoDB.

Configure a real-time synchronization task for a single table

For more information about the procedure, see Configure a real-time synchronization task in Data Integration and Configure a real-time synchronization task in DataStudio.

Configure a synchronization task for an entire database

You can configure tasks for batch synchronization, full and incremental real-time synchronization, or real-time synchronization from sharded databases for an entire database. For more information, see Batch synchronization task for an entire database and Configure a real-time synchronization task for an entire database.

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 sample

The following script is an example of a job that is configured to extract data from MongoDB to a local environment. For more information about the parameters, see the parameter descriptions that follow.

{
    "type":"job",
    "version":"2.0",// Version number.
    "steps":[
        {
            "category": "reader",
            "name": "Reader",
            "parameter": {
                "datasource": "datasourceName", // Data source name.
                "collectionName": "tag_data", // Collection name.
                "query": "", // Data filtering query.
                "column": [
                    {
                        "name": "unique_id", // Field name.
                        "type": "string" // Field type.
                    },
                    {
                        "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" // Time zone.
            } 
        },
        "errorLimit":{
            "record":"0"// Number of error records.
        },
        "speed":{
            "throttle":true,// Specifies whether to enable throttling. If you set this parameter to false, throttling is disabled and the mbps parameter does not take effect. If you set this parameter to true, throttling is enabled.
            "concurrent":1, // Number of concurrent jobs.
            "mbps":"12"// Throttling rate. 1 mbps = 1 MB/s.
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

Writer script sample

The following script is an example of a data synchronization job that is configured to write data to MongoDB. For more information about the parameters, see the parameter descriptions that follow.

{
    "type": "job",
    "version": "2.0",// Version number.
    "steps": [
        {
            "stepType": "stream",
            "parameter": {},
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "mongodb",// Plugin name.
            "parameter": {
                "datasource": "",// Data source name.
                "column": [
                    {
                        "name": "_id",// Column name.
                        "type": "ObjectId"// Data type. If replaceKey is _id, you must set type to ObjectId. If you set 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": {// Write mode.
                    "isReplace": "true",
                    "replaceKey": "_id"
                },
                "collectionName": "datax_test"// Collection name.
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "setting": {
        "errorLimit": {// Number of error records.
            "record": "0"
        },
        "speed": {
            "throttle": true,// Specifies whether to enable throttling. If you set this parameter to false, throttling is disabled and the mbps parameter does not take effect. If you set this parameter to true, throttling is enabled.
            "concurrent": 1,// Number of concurrent jobs.
            "mbps": "1"// Throttling rate. 1 mbps = 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 you run a Data Integration job, the configured preSql is executed first. The actual data writing phase begins only after the preSql execution is complete. The preSql parameter does not affect the content of the data that is written. The preSql parameter provides idempotent execution for Data Integration. For example, your preSql can be used to clear historical data before each task run based on your business rules. In this case, if a task fails, you can simply rerun the Data Integration job.

The format requirements for preSql are as follows:

• You must configure the type field to specify the pre-operation type. The supported values are drop and remove. Example: "preSql":{"type":"remove"}.

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

  • remove: Deletes data based on a condition.

  • json: You can use a JSON object to specify the conditions for data deletion. Example: "preSql":{"type":"remove", "json":"{'operationTime':{'$gte':ISODate('${last_day}T00:00:00.424+0800')}}"}. In this example, ${last_day} is a DataWorks scheduling parameter in the $[yyyy-mm-dd] format. You can also 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 the data deletion query using the following standard MongoDB API.

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

    Note

    To delete data based on conditions, we recommend that you use the JSON configuration.

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

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

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