DataWorks：MongoDB資料來源

對於支援讀取的資料類型，Data Integration在讀取時：

• 基本類型的資料，會根據同步任務配置的讀取欄位（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）預設為英文逗號。

  重要

  Data Integration本身不支援數群組類型，但MongoDB支援數群組類型，並且數群組類型具有強大的索引功能。您可以通過參數的特殊配置，將字串轉換為MongoDB中的數組。轉換類型後，即可並行寫入MongoDB。

結合上文可見，MongoDB Reader針對MongoDB類型的轉換列表，如下表所示。

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

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

配置一個從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":{
        "errorLimit":{
            "record":"0"//錯誤記錄數。
        },
        "speed":{
            "throttle":true,//當throttle值為false時，mbps參數不生效，表示不限流；當throttle值為true時,表示限流。
            "concurrent":1 //作業並發數。
            "mbps":"12"//限流，此處1mbps = 1MB/s。
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

配置寫入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"
            }
        ]
    }
}

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

執行Data Integration作業時，會首先執行您已配置的preSql。完成preSql的執行後，才可以進入實際的資料寫出階段。preSql本身不會影響寫出的資料內容。Data Integration通過preSql參數，可以具備等冪執行特性。例如，您的preSql在每次任務執行前都會清理歷史資料（根據您的商務規則進行清理）。此時，如果任務失敗，您只需要重新執行Data Integration作業即可。

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等）。

    Data Integration通過如下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"}]}。

    Data Integration會基於您配置的item條件項，構造查詢query條件，進而通過MongoDB標準API執行刪除。例如col.deleteMany(query);。

• 不識別的preSql，無需進行任何前置刪除操作。