DataWorks は、Graph Database (GDB) データソースからのデータの読み取りと書き込みを行うための GDB Reader と GDB Writer を提供します。このトピックでは、GDB データソースとの間でデータを同期する機能について説明します。
制限事項
バッチデータ読み取り | バッチデータ書き込み |
|
|
データソースの追加
DataWorks で同期タスクを開発する前に、「データソース管理」の手順に従って、必要なデータソースを DataWorks に追加する必要があります。データソースを追加する際に、DataWorks コンソールでパラメーターの説明を表示して、各パラメーターの意味を理解できます。
データ同期タスクの開発
同期タスクの設定のエントリポイントと手順については、以下の設定ガイドをご参照ください。
単一テーブルのデータを同期するバッチ同期タスクの設定
設定手順の詳細については、「コードレス UI を使用したバッチ同期タスクの設定」および「コードエディタを使用したバッチ同期タスクの設定」をご参照ください。
コードエディタを使用してバッチ同期タスクを設定する際に設定されるすべてのパラメーターと実行されるコードについては、「付録：コードとパラメーター」をご参照ください。
付録：コードとパラメーター
コードエディタを使用したバッチ同期タスクの設定
コードエディタを使用してバッチ同期タスクを設定する場合、統一されたスクリプト形式の要件に基づいて、スクリプト内の関連パラメーターを設定する必要があります。詳細については、「コードエディタでの設定」をご参照ください。以下では、コードエディタを使用してバッチ同期タスクを設定する際に、データソースに対して設定する必要があるパラメーターについて説明します。
Reader スクリプトのデモ
データ同期ジョブを設定して Graph Database (GDB) にデータを書き込む場合、頂点とエッジを個別に設定する必要があります：
GDB インスタンスから頂点に関するデータを読み取る同期タスクの設定
{ "order":{ "hops":[ { "from":"Reader", "to":"Writer" } ] }, "setting":{ "errorLimit":{ "record":"100" // 許容されるダーティデータレコードの最大数。 }, "jvmOption":"", "speed":{ "concurrent":3, "throttle":true, // true の場合、速度制限が有効になります。false の場合、速度制限は無効になり、mbps パラメーターは無視されます。 "mbps":"12"// 速度制限の上限。1 mbps は 1 MB/s に相当します。 } }, "steps":[ { "category":"reader", "name":"Reader", "parameter":{ "host": "gdb-xxxxxx.aliyuncs.com", // GDB インスタンスのエンドポイント。 "port": 8182, // GDB インスタンスのポート。 "username": "gdb", // GDB インスタンスにアクセスするためのユーザー名。 "password": "gdb", // ユーザー名のパスワード。 "labelType": "VERTEX", // ラベルのタイプ。VERTEX は頂点を指定します。 "labels": ["label1", "label2"], // ラベルのリスト。空のままにすると、すべての頂点がエクスポートされます。 "column": [ { "name": "id", // フィールド名。 "type": "string", // フィールドの型。 "columnType": "primaryKey" // フィールドのカテゴリ。頂点のプライマリキーを指定します。GDB では STRING 型である必要があります。 }, { "name": "label", // フィールド名。 "type": "string", // フィールドの型。 "columnType": "primaryLabel" // フィールドのカテゴリ。頂点のラベルを指定します。GDB では STRING 型である必要があります。 }, { "name": "age", // プロパティ名。 "type": "int", // プロパティの型。 "columnType": "vertexProperty" // フィールドのカテゴリ。GDB の頂点の基本プロパティを指定します。 } ] }, "stepType":"gdb" }, { "category":"writer", "name":"Writer", "parameter":{ "print": true }, "stepType":"stream" } ] }GDB インスタンスからエッジに関するデータを読み取る同期タスクの設定
{ "order":{ "hops":[ { "from":"Reader", "to":"Writer" } ] }, "setting":{ "errorLimit":{ "record":"100" // 許容されるダーティデータレコードの最大数。 }, "jvmOption":"", "speed":{ "concurrent":3, "throttle":true,// true の場合、速度制限が有効になります。false の場合、速度制限は無効になり、mbps パラメーターは無視されます。 "mbps":"12"// 速度制限の上限。1 mbps は 1 MB/s に相当します。 } }, "steps":[ { "category":"reader", "name":"Reader", "parameter":{ "host": "gdb-xxxxxx.aliyuncs.com", // GDB インスタンスのエンドポイント。 "port": 8182, // GDB インスタンスのポート。 "username": "gdb", // GDB インスタンスにアクセスするためのユーザー名。 "password": "gdb", // ユーザー名のパスワード。 "labelType": "EDGE", // ラベルのタイプ。EDGE はエッジを指定します。 "labels": ["label1", "label2"], // ラベルのリスト。空のままにすると、すべてのエッジがエクスポートされます。 "column": [ { "name": "id", // フィールド名。 "type": "string", // フィールドの型。 "columnType": "primaryKey" // フィールドのカテゴリ。エッジのプライマリキーを指定します。GDB では STRING 型である必要があります。 }, { "name": "label", // フィールド名。 "type": "string", // フィールドの型。 "columnType": "primaryLabel" // フィールドのカテゴリ。エッジのラベルを指定します。GDB では STRING 型である必要があります。 }, { "name": "srcId", // フィールド名。 "type": "string", // フィールドの型。 "columnType": "srcPrimaryKey" // フィールドのカテゴリ。開始頂点のプライマリキーを指定します。GDB では STRING 型である必要があります。 }, { "name": "srcLabel", // フィールド名。 "type": "string", // フィールドの型。 "columnType": "srcPrimaryLabel" // フィールドのカテゴリ。開始頂点のラベルを指定します。GDB では STRING 型である必要があります。 }, { "name": "dstId", // フィールド名。 "type": "string", // フィールドの型。 "columnType": "dstPrimaryKey" // フィールドのカテゴリ。終了頂点のプライマリキーを指定します。GDB では STRING 型である必要があります。 }, { "name": "dstLabel", // フィールド名。 "type": "string", // フィールドの型。 "columnType": "dstPrimaryLabel" // フィールドのカテゴリ。終了頂点のラベルを指定します。GDB では STRING 型である必要があります。 }, { "name": "weight", // プロパティ名。 "type": "double", // プロパティの型。 "columnType": "edgeProperty" // フィールドのカテゴリ。エッジのプロパティを指定します。 } ] }, "stepType":"gdb" }, { "category":"writer", "name":"Writer", "parameter":{ "print": true }, "stepType":"stream" } ] }
Reader スクリプトのパラメーター
パラメーター | 説明 | 必須 | デフォルト値 |
host | GDB インスタンスの接続エンドポイント。GDB コンソールで、ターゲットインスタンスの Management をクリックして、Internal Endpoint (ホスト) を表示します。 | はい | デフォルト値なし |
port | GDB インスタンスへの接続に使用されるポート番号。 | はい | 8182 |
username | GDB インスタンスへの接続に使用されるユーザー名。 | はい | デフォルト値なし |
password | GDB インスタンスへの接続に使用されるパスワード。 | はい | デフォルト値なし |
labels | ラベル。頂点またはエッジの名前です。GDB Reader は一度に複数の頂点またはエッジからデータを読み取ることができます。この場合、このパラメーターの値は ["label1", "label2"] のような配列になります。 | はい | デフォルト値なし |
labelType | ラベルのタイプ。有効値：
| はい | デフォルト値なし |
column | 同期する頂点またはエッジ。 | はい | デフォルト値なし |
column -> name | 同期する頂点またはエッジのプロパティの名前。頂点またはエッジのプロパティを同期する場合、このパラメーターは必須です。 | はい | デフォルト値なし |
column -> type | 同期する頂点またはエッジのプロパティを格納するためのデータ型。
| はい | デフォルト値なし |
column -> columnType | 同期する頂点またはエッジのプロパティのカテゴリ。
| はい | デフォルト値なし |
Writer スクリプトのデモ
頂点に関するデータを GDB データベースに書き込む同期タスクの設定
{ "order":{ "hops":[ { "from":"Reader", "to":"Writer" } ] }, "setting":{ "errorLimit":{ "record":"100" // 許容されるダーティデータレコードの最大数。 }, "speed":{ "throttle":true,// true の場合、速度制限が有効になります。false の場合、速度制限は無効になり、mbps パラメーターは無視されます。 "concurrent":3, // ジョブの同時実行数。 "mbps":"12"// 速度制限の上限。1 mbps は 1 MB/s に相当します。 } }, "steps":[ { "category":"reader", "name":"Reader", "parameter":{ "column":[ "*" ], "datasource":"_ODPS", "emptyAsNull":true, "guid":"", "isCompress":false, "partition":[], "table":"" }, "stepType":"odps" }, { "category":"writer", "name":"Writer", "parameter": { "datasource": "testGDB", // データソースの名前。 "label": "person", // ラベル。頂点の名前です。 "srcLabel": "", // 頂点には適用されません。 "dstLabel": "", // 頂点には適用されません。 "labelType": "VERTEX", // ラベルのタイプ。「VERTEX」は頂点を指定します。 "writeMode": "INSERT", // 重複するプライマリキーを処理するためのポリシー。 "idTransRule": "labelPrefix", // 頂点のプライマリキーの変換ルール。 "srcIdTransRule": "none", // 頂点には適用されません。 "dstIdTransRule": "none", // 頂点には適用されません。 "column": [ { "name": "id", // フィールド名。 "value": "#{0}", // インデックス 0 のソース列から値を取得します。連結がサポートされています。 "type": "string", // フィールドの型。 "columnType": "primaryKey" // フィールドのカテゴリ。`primaryKey` はプライマリキーを指定します。 }, // 頂点のプライマリキー。フィールド名は `id`、型は STRING である必要があり、このレコードは必須です。 { "name": "person_age", "value": "#{1}", // インデックス 1 のソース列から値を取得します。連結がサポートされています。 "type": "int", "columnType": "vertexProperty" // フィールドのカテゴリ。`vertexProperty` は頂点のプロパティを指定します。 }, // 頂点のプロパティ。サポートされている型：INT、LONG、FLOAT、DOUBLE、BOOLEAN、STRING。 { "name": "person_credit", "value": "#{2}", // インデックス 2 のソース列から値を取得します。連結がサポートされています。 "type": "string", "columnType": "vertexProperty" } // 頂点のプロパティ。 ] } "stepType":"gdb" } ], "type":"job", "version":"2.0" }エッジに関するデータを GDB データベースに書き込む同期タスクの設定
{ "order":{ "hops":[ { "from":"Reader", "to":"Writer" } ] }, "setting":{ "errorLimit":{ "record":"100" // 許容されるダーティデータレコードの最大数。 }, "jvmOption":"", "speed":{ "throttle":true,// true の場合、速度制限が有効になります。false の場合、速度制限は無効になり、mbps パラメーターは無視されます。 "concurrent":3, // ジョブの同時実行数。 "mbps":"12"// 速度制限の上限。1 mbps は 1 MB/s に相当します。 } }, "steps":[ { "category":"reader", "name":"Reader", "parameter":{ "column":[ "*" ], "datasource":"_ODPS", "emptyAsNull":true, "guid":"", "isCompress":false, "partition":[], "table":"" }, "stepType":"odps" }, { "category":"writer", "name":"Writer", "parameter": { "datasource": "testGDB", // データソースの名前。 "label": "use", // ラベル。エッジの名前です。 "labelType": "EDGE", // ラベルのタイプ。`EDGE` はエッジを指定します。 "srcLabel": "person", // 開始頂点のラベル。 "dstLabel": "software", // 終了頂点のラベル。 "writeMode": "INSERT", // 重複するプライマリキーを処理するためのポリシー。 "idTransRule": "labelPrefix", // エッジのプライマリキーの変換ルール。 "srcIdTransRule": "labelPrefix", // 開始頂点のプライマリキーの変換ルール。 "dstIdTransRule": "labelPrefix", // 終了頂点のプライマリキーの変換ルール。 "column": [ { "name": "id", // フィールド名。 "value": "#{0}", // インデックス 0 のソース列から値を取得します。連結がサポートされています。 "type": "string", // フィールドの型。 "columnType": "primaryKey" // フィールドのカテゴリ。`primaryKey` はプライマリキーを指定します。 }, // エッジのプライマリキー。フィールド名は `id`、型は STRING である必要があります。このレコードはオプションです。 { "name": "id", "value": "#{1}", // 連結がサポートされています。マッピングルールが頂点のインポート時に使用されたものと一致していることを確認してください。 "type": "string", "columnType": "srcPrimaryKey" // フィールドのカテゴリ。`srcPrimaryKey` は開始頂点のプライマリキーを指定します。 }, // 開始頂点のプライマリキー。フィールド名は `id`、型は STRING である必要があり、このレコードは必須です。 { "name": "id", "value": "#{2}", // 連結がサポートされています。マッピングルールが頂点のインポート時に使用されたものと一致していることを確認してください。 "type": "string", "columnType": "dstPrimaryKey" // フィールドのカテゴリ。`dstPrimaryKey` は終了頂点のプライマリキーを指定します。 }, // 終了頂点のプライマリキー。フィールド名は `id`、型は STRING である必要があり、このレコードは必須です。 { "name": "person_use_software_time", "value": "#{3}", // 連結がサポートされています。 "type": "long", "columnType": "edgeProperty" // フィールドのカテゴリ。`edgeProperty` はエッジのプロパティを指定します。 }, // エッジのプロパティ。サポートされている型：INT、LONG、FLOAT、DOUBLE、BOOLEAN、STRING。 { "name": "person_regist_software_name", "value": "#{4}", // 連結がサポートされています。 "type": "string", "columnType": "edgeProperty" }, // エッジのプロパティ。 { "name": "id", "value": "#{5}", // 連結がサポートされています。 "type": "long", "columnType": "edgeProperty" } // フィールド名が `id` のエッジのプロパティ。これは通常のプロパティであり、プライマリキーではなく、オプションです。 ] } "stepType":"gdb" } ], "type":"job", "version":"2.0" }
Writer スクリプトのパラメーター
パラメーター | 説明 | 必須 | デフォルト値 |
datasource | データソースの名前。追加されたデータソースの名前と同じである必要があります。コードエディタを使用してデータソースを追加できます。 | はい | デフォルト値なし |
label | ラベル。頂点またはエッジの名前です。 GDB Writer はソーステーブルの列からラベルを取得できます。たとえば、このパラメーターを #{0} に設定すると、GDB Writer は最初の列の値をラベルとして使用します。列のインデックスは 0 から始まります。 | はい | デフォルト値なし |
labelType | ラベルのタイプ。有効値：
| はい | デフォルト値なし |
srcLabel |
| いいえ | デフォルト値なし |
dstLabel |
| いいえ | デフォルト値なし |
writeMode | GDB Writer が重複するプライマリキーを持つデータレコードを処理するモード。有効値：
| はい | INSERT |
idTransRule | プライマリキーを変換するためのルール。有効値：
| はい | none |
srcIdTransRule | labelType パラメーターが EDGE に設定されている場合の、開始頂点のプライマリキーを変換するためのルール。有効値：
| labelType パラメーターが EDGE に設定されている場合は必須 | none |
dstIdTransRule | labelType パラメーターが EDGE に設定されている場合の、終了頂点のプライマリキーを変換するためのルール。有効値：
| labelType パラメーターが EDGE に設定されている場合は必須 | none |
column | 同期したい頂点またはエッジ。
プロパティのサンプル | はい | デフォルト値なし |