配置一個(gè)從MongoDB抽取數(shù)據(jù)到本地的作業(yè)，詳情請(qǐng)參見下文的參數(shù)說(shuō)明。

{
    "type":"job",
    "version":"2.0",//版本號(hào)。
    "steps":[
        {
            "category": "reader",
            "name": "Reader",
            "parameter": {
                "datasource": "datasourceName", //數(shù)據(jù)源名稱。
                "collectionName": "tag_data", //集合名稱。
                "query": "", // 數(shù)據(jù)查詢過(guò)濾。
                "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":{
        "common": { 
            "column": { 
                "timeZone": "GMT+0" //時(shí)區(qū)
            } 
        },
        "errorLimit":{
            "record":"0"http://錯(cuò)誤記錄數(shù)。
        },
        "speed":{
            "throttle":true,//當(dāng)throttle值為false時(shí)，mbps參數(shù)不生效，表示不限流；當(dāng)throttle值為true時(shí),表示限流。
            "concurrent":1 //作業(yè)并發(fā)數(shù)。
            "mbps":"12"http://限流，此處1mbps = 1MB/s。
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

配置寫入MongoDB的數(shù)據(jù)同步作業(yè)，詳情請(qǐng)參見下文的參數(shù)說(shuō)明。

{
    "type": "job",
    "version": "2.0",//版本號(hào)。
    "steps": [
        {
            "stepType": "stream",
            "parameter": {},
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "mongodb",//插件名。
            "parameter": {
                "datasource": "",//數(shù)據(jù)源名。
                "column": [
                    {
                        "name": "_id",//列名。
                        "type": "ObjectId"http://數(shù)據(jù)類型。如果replacekey為_id，則此處的type必須配置為ObjectID。如果配置為string，會(huì)無(wú)法進(jìn)行替換。
                    },
                    {
                        "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"http://連接名稱。
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "setting": {
        "errorLimit": {//錯(cuò)誤記錄數(shù)。
            "record": "0"
        },
        "speed": {
            "throttle": true,//當(dāng)throttle值為false時(shí)，mbps參數(shù)不生效，表示不限流；當(dāng)throttle值為true時(shí),表示限流。
            "concurrent": 1,//作業(yè)并發(fā)數(shù)。
            "mbps": "1"http://限流的速度，此處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"

執(zhí)行數(shù)據(jù)集成作業(yè)時(shí)，會(huì)首先執(zhí)行您已配置的preSql。完成preSql的執(zhí)行后，才可以進(jìn)入實(shí)際的數(shù)據(jù)寫出階段。preSql本身不會(huì)影響寫出的數(shù)據(jù)內(nèi)容。數(shù)據(jù)集成通過(guò)preSql參數(shù)，可以具備冪等執(zhí)行特性。例如，您的preSql在每次任務(wù)執(zhí)行前都會(huì)清理歷史數(shù)據(jù)（根據(jù)您的業(yè)務(wù)規(guī)則進(jìn)行清理）。此時(shí)，如果任務(wù)失敗，您只需要重新執(zhí)行數(shù)據(jù)集成作業(yè)即可。

preSql的格式要求如下：

• 需要配置type字段，表示前置操作類別，支持drop和remove，例如"preSql":{"type":"remove"}：

  • drop：表示刪除集合和集合內(nèi)的數(shù)據(jù)，collectionName參數(shù)配置的集合即是待刪除的集合。

  • remove：表示根據(jù)條件刪除數(shù)據(jù)。

  • json：您可以通過(guò)JSON控制待刪除的數(shù)據(jù)條件，例如"preSql":{"type":"remove", "json":"{'operationTime':{'$gte':ISODate('${last_day}T00:00:00.424+0800')}}"}。此處的${last_day}為DataWorks調(diào)度參數(shù)，格式為$[yyyy-mm-dd]。您可以根據(jù)需要具體使用其它MongoDB支持的條件操作符號(hào)（$gt、$lt、$gte和$lte等）、邏輯操作符（and和or等）或函數(shù)（max、min、sum、avg和ISODate等）。

    數(shù)據(jù)集成通過(guò)如下MongoDB標(biāo)準(zhǔn)API執(zhí)行您的數(shù)據(jù)，刪除query。

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

    說(shuō)明

    如果您需要條件刪除數(shù)據(jù)，建議優(yōu)先使用JSON配置形式。

  • item：您可以在item中配置數(shù)據(jù)過(guò)濾的列名（name）、條件（condition）和列值（value）。例如"preSql":{"type":"remove","item":[{"name":"pv","value":"100","condition":"$gt"},{"name":"pid","value":"10"}]}。

    數(shù)據(jù)集成會(huì)基于您配置的item條件項(xiàng)，構(gòu)造查詢query條件，進(jìn)而通過(guò)MongoDB標(biāo)準(zhǔn)API執(zhí)行刪除。例如col.deleteMany(query);。

• 不識(shí)別的preSql，無(wú)需進(jìn)行任何前置刪除操作。