DataWorks:Use a batch synchronization task to read data from a MongoDB data source

Prerequisites

Make preparations

In this example, you must prepare a MongoDB data collection and a MaxCompute table for data synchronization.

1. Prepare a MongoDB data collection.

   In this example, ApsaraDB for MongoDB is used. The following sample code provides an example on how to prepare an ApsaraDB for MongoDB data collection.

   1. Create a data collection named di_mongodb_conf_test.

      db.createCollection('di_mongodb_conf_test')

   2. Insert sample data into the data collection.

      db.di_mongodb_conf_test.insertOne({
          'col_string':'mock string value',
          'col_int32':NumberInt("1"),
          'col_int32_min':NumberInt("-2147483648"),
          'col_int32_max':NumberInt("2147483647"),
          'col_int64':NumberLong("1234567890123456"),
          'col_int64_min':NumberLong("-9223372036854775807"),
          'col_int64_max':NumberLong("9223372036854775807"),
          'col_decimal':NumberDecimal("9999999.4999999999"),
          'col_double':9999999.99,
          'col_boolean':true,
          'col_timestamp':ISODate(),
          'col_date':new Date(),
          'col_array_to_json':['a','b'],
          'col_array_to_join':['a','b'],
          'col_doc':{
              'key_string':'mock string value',
              'key_int32':NumberInt("1"),
              'key_int32_min':NumberInt("-2147483648"),
              'key_int32_max':NumberInt("2147483647"),
              'key_int64':NumberLong("1234567890123456"),
              'key_int64_min':NumberLong("-9223372036854775807"),
              'key_int64_max':NumberLong("9223372036854775807"),
              'key_decimal':NumberDecimal("9999999.4999999999"),
              'key_double':9999999.99,
              'key_boolean':true,
              'key_timestamp':ISODate(),
              'key_date':new Date(),
              'key_array_to_json':['a','b'],
              'key_array_to_join':['a','b'],
          },
          'col_extra_1':'this is extra 1',
          'col_extra_2':'this is extra 2',
      })

   3. Query the data that is inserted into the MongoDB data collection.

      db.getCollection("di_mongodb_conf_test").find({})

      The following figure shows the query result.

2. Prepare a MaxCompute table.

   1. Create a partitioned table named di_mongodb_conf_test. The partition field is pt.

      CREATE TABLE IF NOT EXISTS di_mongodb_conf_test
      (
        `id`                 STRING
        ,`col_string`        STRING
        ,`col_int32`         INT
        ,`col_int32_min`     INT
        ,`col_int32_max`     INT
        ,`col_int64`         BIGINT
        ,`col_int64_min`     BIGINT
        ,`col_int64_max`     BIGINT
        ,`col_decimal`       DECIMAL(38,18)
        ,`col_double`        DOUBLE
        ,`col_boolean`       BOOLEAN
        ,`col_timestamp`     TIMESTAMP
        ,`col_date`          DATE
        ,`col_array_to_json` STRING
        ,`col_array_to_join` STRING
        ,`key_string`        STRING
        ,`key_int32`         INT
        ,`key_int32_min`     INT
        ,`key_int32_max`     INT
        ,`key_int64`         BIGINT
        ,`key_int64_min`     BIGINT
        ,`key_int64_max`     BIGINT
        ,`key_decimal`       DECIMAL(38,18)
        ,`key_double`        DOUBLE
        ,`key_boolean`       BOOLEAN
        ,`key_timestamp`     TIMESTAMP
        ,`key_date`          DATE
        ,`key_array_to_json` STRING
        ,`key_array_to_join` STRING
        ,`col_doc`           STRING
        ,`col_combine`       STRING
      )
      PARTITIONED BY
      (
        pt                   STRING
      )
      LIFECYCLE 36500
      ;

   2. Add the value 20230202 to the partition field.

      alter table di_mongodb_conf_test add if not exists partition (pt='20230202');

   3. Check whether the partitioned table is correctly created.

      SELECT*FROM di_mongodb_conf_test
      WHEREpt='20230202';

Configure a batch synchronization task

Step 1: Add a MongoDB data source

Add a MongoDB data source and make sure that a network connection is established between the data source and the exclusive resource group for Data Integration. For more information, see Add a MongoDB data source.

Step 2: Create and configure a batch synchronization task

Step 3: Commit and deploy the batch synchronization task

If you use a workspace in standard mode and you want to periodically schedule the batch synchronization task in the production environment, you can commit and deploy the task to the production environment. For more information, see Deploy nodes.

Step 4: Run the batch synchronization task and view the synchronization result

Appendix 1: Data type conversion during data synchronization

Convert arrays into JSON data: col_array_to_json

{
    "col_array_to_json":
    [
        "a",
        "b"
    ]
}

{"name":"col_array_to_json","type":"string"}

[a, b]

Convert arrays into concatenated strings: col_array_to_join

{
    "col_array_to_join":
    [
        "a",
        "b"
    ]
}

{"name":"col_array_to_join","type":"array","splitter":","}

a,b

Read data from a specified nested field in a MongoDB document

{
    "col_doc":
    {
        "key_string": "mock string value"
    }
}

{"name":"col_doc.key_string","type":"document.string"}

mock string value

Convert data in a MongoDB document into JSON data

{
    "col_doc":
    {
        "key_string": "mock string value",
        "key_int32": 1
    }
}

{"name":"col_doc","type":"string"}

{"key_string":"mockstringvalue","key_int32":1}

Convert data in fields except the configured fields in a MongoDB document into JSON data

{
    "col_1": "value1",
    "col_2": "value2",
    "col_3": "value3",
    "col_4": "value4"
}

{"name":"col_1","type":"string"}
{"name":"col_2","type":"string"}
{"name":"col_combine","type":"combine"}

{"col_3":"value3","col_4":"value4"}

Appendix 2: Issues related to output data that is converted from data of the DECIMAL type

{
    "key_decimal":
    {
        "finite": true,
        "high": 3471149412795809792,
        "infinite": false,
        "low": 99999994999999999,
        "naN": false,
        "negative": false
    }
}