DataWorks:MongoDB データソース

MongoDB Reader がサポートする MongoDB のデータ型

データ統合は、すべての MongoDB データ型をサポートしているわけではありませんが、ほとんどのデータ型をサポートしています。ご使用のデータ型がサポートされていることを確認してください。

データ統合がサポートされているデータ型を読み取る際、次の操作を実行します：

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

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

• 一部の複雑なデータ型については、列の type プロパティを構成してカスタム処理を実行できます。

  タイプ	バッチ読み取り (MongoDB Reader)	説明
  Document	サポート	埋め込みドキュメント型です。 type プロパティが構成されていない場合、Document は JSON シリアル化を使用して直接変換されます。 type プロパティが document に設定されている場合、フィールドはネストされた型になります。MongoDB Reader はパスに基づいて Document のプロパティを読み取ります。詳細な例については、後述の「例 2：多階層のネストされた Document の再帰的な解析」をご参照ください。
  Array	サポート	配列型です。 type が array.json または arrays に設定されている場合、データは JSON シリアル化を使用して直接処理されます。 type が array または document.array に設定されている場合、要素は文字列に連結されます。区切り文字は、列の splitter プロパティで指定され、デフォルトではカンマ (,) です。

  重要

  データ統合は配列型をサポートしていません。ただし、MongoDB は配列型をサポートしており、強力なインデックス機能を提供しています。特定のパラメーターを構成して、文字列を MongoDB の配列に変換できます。変換後、データを MongoDB に並列で書き込むことができます。

データ統合の特殊なデータ型：combine

MongoDB Reader のデータ型マッピング

次の表に、MongoDB Reader の MongoDB データ型とデータ統合データ型のマッピングを示します。

MongoDB Writer のデータ型マッピング

例 1：combine 型の使用

MongoDB Reader プラグインの combine データ型を使用すると、MongoDB ドキュメント内の複数のフィールドを単一の JSON 文字列にマージできます。たとえば、3 つの MongoDB ドキュメントから MaxCompute にフィールドをインポートするとします。次の例では、フィールドはキーと値のペアではなくキーで表されます。フィールド a と b は 3 つのドキュメントすべてに共通しており、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 のマッピングが必要なフィールドを明示的に指定する必要があります。マージしたいフィールドには、ドキュメント内の既存のフィールド名とは異なる新しい名前を割り当て、type を COMBINE に設定します。次のコードに例を示します。

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

次の表に、MaxCompute での最終的な出力を示します。

例 2：多階層のネストされた Document の再帰的な解析

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 のサンプルスクリプトとパラメーターの説明」をご参照ください。

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

手順の詳細については、「データ統合でのリアルタイム同期タスクの構成」および「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,// スロットルを有効にするかどうかを指定します。このパラメーターを false に設定すると、スロットルは無効になり、mbps パラメーターは効果がありません。このパラメーターを 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 の場合、type を ObjectId に設定する必要があります。type を 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,// スロットルを有効にするかどうかを指定します。このパラメーターを false に設定すると、スロットルは無効になり、mbps パラメーターは効果がありません。このパラメーターを 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"

データ統合ジョブを実行すると、構成された preSql が最初に実行されます。実際のデータ書き込みフェーズは、preSql の実行が完了した後にのみ開始されます。preSql パラメーターは、書き込まれるデータの内容には影響しません。preSql パラメーターは、データ統合にべき等実行を提供します。たとえば、ビジネスルールに基づいて各タスク実行前に既存データをクリアするために preSql を使用できます。この場合、タスクが失敗しても、データ統合ジョブを再実行するだけで済みます。

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) も使用できます。

    データ統合は、次の標準 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"}]}。

    データ統合は、構成された item 条件に基づいてクエリ条件を構築し、標準の MongoDB API を使用して削除を実行します。例：col.deleteMany(query);。

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