DataWorks:MongoDB データソース

MongoDB Reader でサポートされる MongoDB データ型

Data Integration はほとんどの MongoDB データ型をサポートしていますが、一部サポートされていないデータ型もあります。続行する前に、ご利用のデータ型を確認してください。

サポートされているデータ型の場合、Data Integration は次のようにデータを読み取ります：

• 基本データ型の場合、Data Integration は同期タスクに設定された読み取りフィールド (column) の名前に基づいて、対応するパスからデータを自動的に読み取ります。また、データ型も自動的に変換します。カラムに type プロパティを指定する必要はありません。詳細については、「付録：MongoDB スクリプトのデモとパラメーターの説明」をご参照ください。

  タイプ	オフライン読み取り (MongoDB Reader)	説明
  ObjectId	サポート	オブジェクト ID 型です。
  Double	サポート	64 ビット浮動小数点数型です。
  32 ビット整数	サポート	32 ビット整数です。
  64 ビット整数	サポート	64 ビット整数です。
  Decimal128	サポート	Decimal128 型です。 説明 この型がネストされた型または Combine 型として設定されている場合、JSON シリアル化中にオブジェクトとして処理されます。データを 10 進数として出力するには、decimal128OutputType パラメーターを追加し、bigDecimal に設定する必要があります。
  String	サポート	文字列型です。
  Boolean	サポート	ブール値型です。
  Timestamp	サポート	タイムスタンプ型です。 説明 BsonTimestamp はタイムスタンプを格納します。タイムゾーンの影響を考慮する必要はありません。詳細については、「MongoDB のタイムゾーンの問題」をご参照ください。
  Date	サポート	日付型です。

• 一部の複雑なデータ型については、カラムの type プロパティを設定することで、処理をカスタマイズできます。

  タイプ	オフライン読み取り (MongoDB Reader)	説明
  Document	サポート	埋め込みドキュメント型です。 type プロパティが設定されていない場合、ドキュメントは直接 JSON シリアル化によって処理されます。 type プロパティが document に設定されている場合、それはネストされた型です。MongoDB Reader はパスによってドキュメントのプロパティを読み取ります。詳細な例については、「データ型の例 2：複数レベルのネストされたドキュメントの再帰的解析」をご参照ください。
  Array	サポート	配列型です。 type が array.json または arrays に設定されている場合、データは直接 JSON シリアル化によって処理されます。 type が array または document.array に設定されている場合、データは文字列に連結されます。デフォルトの区切り文字 (カラム内の splitter) はカンマ (,) です。

  重要

  Data Integration は配列型をサポートしていません。しかし、MongoDB は配列とその強力なインデックス機能をサポートしています。特定のパラメーターを設定して、MongoDB で文字列を配列に変換できます。変換後、データを MongoDB に並行して書き込むことができます。

特殊な Data Integration データ型：combine

MongoDB Reader のデータ型変換

次の表に、MongoDB Reader が実行するデータ型変換を示します。

MongoDB Writer のデータ型変換

データ型の例 1：combine 型の使用

MongoDB Reader プラグインの Combine データ型を使用すると、MongoDB ドキュメント内の複数のフィールドを単一の JSON 文字列にマージできます。たとえば、MongoDB から MaxCompute にフィールドをインポートしたいとします。次の 3 つのドキュメントには、キーがフィールド全体を表し、値が省略されているフィールドが含まれています。フィールド `a` と `b` はすべてのドキュメントに共通であり、`x_n` は固定されていないフィールドです。

• doc1: a b x_1 x_2

• doc2: a b x_2 x_3 x_4

• doc3: a b x_5

設定ファイルで、1 対 1 のマッピングが必要なフィールドを指定します。マージするフィールドには、ドキュメント内の既存のフィールドとは異なる新しい名前を割り当て、型を COMBINE に設定します。以下に示します。

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

MaxCompute でのエクスポート結果は次のようになります。

データ型の例 2：複数レベルのネストされたドキュメントの再帰的解析

MongoDB ドキュメントに複数レベルのネストがある場合、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 です。

単一テーブルのオフライン同期タスクの設定

• 手順については、「コードレス UI でのタスク設定」および「コードエディタでのタスク設定」をご参照ください。

• コードエディタでタスクを設定するためのすべてのパラメーターとスクリプトのデモについては、「付録：MongoDB スクリプトのデモとパラメーターの説明」をご参照ください。

単一テーブルのリアルタイム同期タスクの設定

手順については、「Data Integration でのリアルタイム同期タスクの設定」および「DataStudio でのリアルタイム同期タスクの設定」をご参照ください。

データベース全体の同期タスクの設定

データベース全体のオフライン同期、データベース全体のフルおよび増分リアルタイム同期、シャーディングを伴うデータベース全体のリアルタイム同期など、データベース全体の同期タスクの設定方法については、「データベース全体のオフライン同期タスク」および「データベース全体のリアルタイム同期タスクの設定」をご参照ください。

コードエディタを使用したバッチ同期タスクの設定

コードエディタを使用してバッチ同期タスクを設定する場合、統一されたスクリプト形式の要件に基づいて、スクリプトに関連パラメーターを設定する必要があります。詳細については、「コードエディタでのタスク設定」をご参照ください。以下の情報は、コードエディタを使用してバッチ同期タスクを設定する際に、データソースに対して設定する必要があるパラメーターについて説明しています。

Reader スクリプトのデモ

次のコードは、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":{
        "common": { 
            "column": { 
                "timeZone": "GMT+0" // タイムゾーン。
            } 
        },
        "errorLimit":{
            "record":"0"// エラーレコードの数。
        },
        "speed":{
            "throttle":true,// throttle が false に設定されている場合、mbps パラメーターは効果がなく、データレートは制限されません。throttle が true に設定されている場合、データレートは制限されます。
            "concurrent":1, // 同時実行ジョブの数。
            "mbps":"12"// データレートの制限。1 mbps は 1 MB/s に相当します。
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

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 に設定されている場合、型は 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"// データレートの制限。1 mbps は 1 MB/s に相当します。
        },
       "jvmOption": "-Xms1024m -Xmx1024m"
    },
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    }
}

Writer スクリプトのパラメーター

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

Data Integration ジョブが実行されると、設定された preSql が最初に実行されます。実際のデータ書き込みフェーズは、preSql が完了した後にのみ開始されます。preSql 自体は、書き込まれるデータの内容には影響しません。preSql パラメーターにより、Data Integration はべき等実行をサポートできます。たとえば、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} は $[yyyy-mm-dd] 形式の DataWorks スケジューリングパラメーターです。必要に応じて、他の MongoDB がサポートする条件付きオペレーター ($gt、$lt、$gte、$lte など)、論理演算子 (and、or など)、または関数 (max、min、sum、avg、ISODate など) を使用できます。

    Data Integration は、次の標準 MongoDB API を使用して、クエリを削除するためのデータ操作を実行します。

    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 条件に基づいてクエリ条件を構築し、標準の MongoDB API を使用して削除を実行します。例：col.deleteMany(query);。

• preSql が認識されない場合、事前削除操作は実行されません。