離線同步讀取MongoDB資料

前提條件

準備樣本資料表

本實踐需準備一個MongoDB資料集合、一個MaxCompute表，用於後續進行離線資料同步。

1. 準備MongoDB資料集合。

   本實踐使用阿里雲ApsaraDB for MongoDB作為樣本。以下為準備MongoDB資料集合的主要程式碼範例。

   1. 建立一個名稱為di_mongodb_conf_test的資料集合。

      db.createCollection('di_mongodb_conf_test')

   2. 向資料集合中插入本實踐的樣本資料。

      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. 查詢插入MongoDB中的資料。

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

      查詢結果為：查詢結果顯示集合中包含一條測試文檔（_id 為 63dca714b8548a78e1dc3238），文檔涵蓋多種資料類型欄位：col_string（字串）、col_int32/col_int64（整數，含最大/最小值）、col_double/col_decimal（浮點數）、col_boolean（布爾值）、col_date/col_timestamp（日期/時間戳）、col_array_to_join/col_array_to_json（數組）、col_doc（嵌套文檔，內含 key_string、key_int32、key_int64、key_decimal、key_double、key_boolean、key_timestamp、key_date、key_array_to_json、key_array_to_join 等子欄位）以及 col_extra_1、col_extra_2 附加欄位。

2. 準備MaxCompute表。

   1. 建立一個名稱為di_mongodb_conf_test的分區表，分區欄位為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. 添加一個分區取值20230202。

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

   3. 檢查分區表是否正確建立。

      SELECT*FROM di_mongodb_conf_test
      WHEREpt='20230202';

離線任務配置

step1：添加MongoDB資料來源

添加一個MongoDB資料來源，保障資料來源與獨享Data Integration資源群組之間網路連通。操作詳情請參見配置MongoDB資料來源。

step2：建立離線同步節點，並配置離線同步任務

step3：提交發布離線同步節點

如果您使用的是標準模式的DataWorks工作空間，並且希望後續在生產環境中周期性調度此離線同步任務的話，您可以將離線同步節點提交發布到生產環境。操作詳情請參見發布任務。

step4：運行離線同步節點，查看同步結果

附錄1：同步過程中的資料格式轉換說明

數群組類型資料轉換為JSON格式輸出：col_array_to_json

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

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

[a, b]

數群組類型資料轉換為拼接字串格式輸出：col_array_to_join

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

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

a,b

Document資料存在多層嵌套時，讀取嵌套中的指定欄位同步

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

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

mock string value

Document資料作為json序列化輸出

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

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

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

Document資料除去已配置欄位外，其他欄位整體進行json序列化輸出

{
    "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"}

附錄2：關於document中Decimal類型輸出問題

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