DataWorks:Elasticsearch データソース

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

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

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

単一テーブルのリアルタイム書き込みタスクの構成

手順の詳細については、「DataStudio でリアルタイム同期タスクを構成する」をご参照ください。

データベース全体のオフライン書き込みタスク、または単一テーブルまたはデータベース全体の完全/増分リアルタイム書き込みタスクの構成

手順の詳細については、「データベース全体のリアルタイム同期タスクを構成する」をご参照ください。

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

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

Reader スクリプトデモ

{
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    },
    "setting":{
        "errorLimit":{
            "record":"0" // エラーレコードの数。
        },
        "jvmOption":"",
        "speed":{
            "concurrent":3,
            "throttle":false
        }
    },
    "steps":[
        {
            "category":"reader",
            "name":"Reader",
            "parameter":{
                "column":[ // 読み取る列。
                    "id",
                    "name"
                ],
                "endpoint":"http://es-cn-xxx.elasticsearch.aliyuncs.com:9200", // エンドポイント。
                "index":"aliyun_es_xx",  // インデックス。
                "password":"*******",  // パスワード。
                "multiThread":true,
                "scroll":"5m",  // スクロールフラグ。
                "pageSize":5000,
                "connTimeOut":600000,
                "readTimeOut":600000,
                "retryCount":30,
                "retrySleepTime":"10000",
                "search":{
                            "range":{
                                "gmt_modified":{
                                    "gte":0
                                }
                            }
                        },  // クエリパラメーター。これは Elasticsearch のクエリコンテンツと同じです。_search API が使用され、search に名前が変更されます。
                "type":"doc",
                "username":"aliyun_di"  // ユーザー名。
            },
            "stepType":"elasticsearch"
        },
        {
            "category":"writer",
            "name":"Writer",
            "parameter":{ },
            "stepType":"stream"
        }
    ],
    "type":"job",
    "version":"2.0" // バージョン番号。
}

Reader スクリプトパラメーター

Writer スクリプトデモ

{
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    },
    "setting": {
        "errorLimit": {
            "record": "0"
        },
        "speed": {
            "throttle":true,// throttle が false に設定されている場合、mbps パラメーターは効果がなく、スロットリングは行われません。throttle が true に設定されている場合、スロットリングが有効になります。
            "concurrent":1, // 同時実行ジョブの数。
            "mbps":"12"// 速度制限。1 Mbps = 1 MB/s。
        }
    },
    "steps": [
        {
            "category": "reader",
            "name": "Reader",
            "parameter": {

            },
            "stepType": "stream"
        },
        {
            "category": "writer",
            "name": "Writer",
            "parameter": {
                "datasource":"xxx",
                "index": "test-1",
                "type": "default",
                "cleanup": true,
                "settings": {
                        "number_of_shards": 1,
                        "number_of_replicas": 0
                },
                "discovery": false,
                "primaryKeyInfo":{
                    "type":"pk",    
                     "fieldDelimiter":",",
                     "column":[]
                    },
                "batchSize": 1000,
                "dynamic":false,
                "esPartitionColumn":[
                    {
                        "name":"col1",  
                        "comment":"xx", 
                        "type":"STRING" 
                        }
                     ],
                "column": [
                    {
                        "name": "pk",
                        "type": "id"
                    },
                    {
                        "name": "col_ip",
                        "type": "ip"
                    },
                    {
                        "name": "col_array",
                        "type": "long",
                        "array": true
                    },
                    {
                        "name": "col_double",
                        "type": "double"
                    },
                    {
                        "name": "col_long",
                        "type": "long"
                    },
                    {
                        "name": "col_integer",
                        "type": "integer"
                    },
                    {
                        "name": "col_keyword",
                        "type": "keyword"
                    },
                    {
                        "name": "col_text",
                        "type": "text",
                        "analyzer": "ik_max_word",
                        "other_params":
                            {
                                "doc_values": false
                            }
                    },
                    {
                        "name": "col_geo_point",
                        "type": "geo_point"
                    },
                    {
                        "name": "col_date",
                        "type": "date",
                        "format": "yyyy-MM-dd HH:mm:ss"
                    },
                    {
                        "name": "col_nested1",
                        "type": "nested"
                    },
                    {
                        "name": "col_nested2",
                        "type": "nested"
                    },
                    {
                        "name": "col_object1",
                        "type": "object"
                    },
                    {
                        "name": "col_object2",
                        "type": "object"
                    },
                    {
                        "name": "col_integer_array",
                        "type": "integer",
                        "array": true
                    },
                    {
                        "name": "col_geo_shape",
                        "type": "geo_shape",
                        "tree": "quadtree",
                        "precision": "10m"
                    }
                ]
            },
            "stepType": "elasticsearch"
        }
    ],
    "type": "job",
    "version": "2.0"
}

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

- id  // id タイプは Elasticsearch の _id に対応し、一意のプライマリキーとして理解できます。データを書き込む際、同じ ID のデータは上書きされ、インデックスは作成されません。
- string
- text
- keyword
- long
- integer
- short
- byte
- double
- float
- date
- boolean
- binary
- integer_range
- float_range
- long_range
- double_range
- date_range
- geo_point
- geo_shape
- ip
- token_count
- array
- object
- nested

 {
   "name": "guid",
   "other_params":
    {
       "doc_values": false
      },
    "type": "text"
  }

シナリオ 1: 完全なデータプル

• 背景: Elasticsearch のドキュメントのクエリ結果を単一のフィールドとしてプルできます。

• 例:

  
  ## Reader: Elasticsearch の生データ
  "hits": [
      {
          "_index": "mutiltest_1",
          "_type": "_doc",
          "_id": "IXgdO4MB4GR_1DmrjTXP",
          "_score": 1.0,
          "_source": {
              "feature1": "value1",
              "feature2": "value2",
              "feature3": "value3"
          }
      }]
  
  ## Data Integration Elasticsearch Reader プラグインの構成
  "parameter": {
    "column": [
        "content"
    ],
    "full":true
  }
  
  ## Writer の結果: 1 行 1 列が宛先に同期されます。
  {"_index":"mutiltest_1","_type":"_doc","_id":"IXgdO4MB4GR_1DmrjTXP","_source":{"feature1":"value1","feature2":"value2","feature3":"value3"},"sort":["IXgdO4MB4GR_1DmrjTXP"]}

シナリオ 2: ネストされたフィールドまたはオブジェクトフィールドのプロパティを同期する

• 背景: Object またはネストされたフィールドのプロパティを同期する場合、パスを使用してそれらを解決できます。

• 構成:

  • property

  • property.sub-property

  • property[0].sub-property

• スクリプト構成:

  "multi":{
      "multi":true
  }

  説明

  現在、コードレス UI での構成は利用できません。

• 例:

  ## Reader: Elasticsearch の生データ
  "hits": [
      {
          "_index": "mutiltest_1",
          "_type": "_doc",
          "_id": "7XAOOoMB4GR_1Dmrrust",
          "_score": 1.0,
          "_source": {
              "level1": {
                  "level2": [
                      {
                          "level3": "testlevel3_1"
                      },
                      {
                          "level3": "testlevel3_2"
                      }
                  ]
              }
          }
      }
  ]
  ## Data Integration Elasticsearch reader プラグインの構成
  "parameter": {
    "column": [
        "level1",
        "level1.level2",
        "level1.level2[0]",
        "level1.level2.level3"
    ],
    "multi":{
          "multi":true
      }
  }
  
  ## Writer の結果: 4 列のデータが 1 行
  column1(level1):            {"level2":[{"level3":"testlevel3_1"},{"level3":"testlevel3_2"}]}
  column2(level1.level2):     [{"level3":"testlevel3_1"},{"level3":"testlevel3_2"}]
  column3(level1.level2[0]):  {"level3":"testlevel3_1"}
  column4(level1.level2.level3):  null

  説明

  • 取得したノードの親ノードに配列が含まれている場合、結果は null になります。たとえば、`level1.level2.level3` を取得してもエラーは報告されませんが、同期結果は null になります。パスを `level1.level2[0].level3` または `level1.level2[1].level3` として構成する必要があります。`level1.level2[*].level3` 形式は現在サポートされていません。

  • キーにピリオド (.) を含むデータはサポートされていません。たとえば、`{"level1.level2":{"level3":"testlevel3_1"}}` のようなデータです。この場合、このデータエントリの取得結果は null になります。

シナリオ 3: 配列プロパティを複数の行に分割する

• 背景: 1 対多の関係が存在する場合、配列列を複数の行に分割できます。

• 構成: property[*].sub-property

• 効果: `{ "splitKey" :[1,2,3,4,5]}` のようなソースデータは、5 つの別々の行として分割されて宛先に書き込まれます。

• スクリプト構成:

  "multi":{   
         "multi":true,    
          "key": "headers"
  }

  説明

  • コードレス UI で [多行配列列名の分割] を構成すると、同等のスクリプト構成が自動的に生成されます。

  • 値はリストである必要があります。そうでない場合、エラーが発生します。

• 例:

  ## Reader: Elasticsearch の生データ
  [
      {
          "_index": "lmtestjson",
          "_type": "_doc",
          "_id": "nhxmIYMBKDL4VkVLyXRN",
          "_score": 1.0,
          "_source": {
              "headers": [
                  {
                      "remoteip": "192.0.2.1"
                  },
                  {
                      "remoteip": "192.0.2.2"
                  }
              ]
          }
      },
      {
          "_index": "lmtestjson",
          "_type": "_doc",
          "_id": "wRxsIYMBKDL4VkVLcXqf",
          "_score": 1.0,
          "_source": {
              "headers": [
                  {
                      "remoteip": "192.0.2.3"
                  },
                  {
                      "remoteip": "192.0.2.4"
                  }
              ]
          }
      }
  ]
  ## Data Integration Elasticsearch reader プラグインの構成
  {
     "column":[
        "headers[*].remoteip"
    ]
    "multi":{
        "multi":true,
        "key": "headers"
    }
  }
  
  ## Writer の結果: 4 行
  192.0.2.1
  192.0.2.2
  192.0.2.3
  192.0.2.4

シナリオ 4: 配列プロパティの重複排除とマージ

• 背景: 配列プロパティを重複排除してマージし、その結果を文字列プロパティとして書き込むことができます。配列プロパティは、`name1.name2` のようなサブプロパティにすることができます。`toString` の結果が重複排除の基準として使用されます。

• 構成: `property[]`。

  列名に `[]` が含まれている場合、このプロパティに対して重複排除とマージが実行されます。

• スクリプト構成:

  "multi":{
      "multi":true
  }

  説明

  このパラメーターはコードレス UI では構成できません。

• 例:

  ## Reader: Elasticsearch の生データ
  "hits": [
  {
      "_index": "mutiltest_1",
      "_type": "_doc",
      "_id": "4nbUOoMB4GR_1Dmryj8O",
      "_score": 1.0,
      "_source": {
          "feature1": [
              "value1",
              "value1",
              "value2",
              "value2",
              "value3"
          ]
      }
  }
  ]
  ## Data Integration Elasticsearch reader プラグインの構成
  "parameter": {
    "column":[
          "feature1[]"
    ],
    "multi":{
          "multi":true
      }
  }
  
  ## Writer の結果: 1 行 1 列のデータ
  "value1,value2,value3"

シナリオ 5: 複数のプロパティをマージして同期する

• 背景: 複数のプロパティを選択的に処理できます。値を持つ最初のプロパティが返されます。指定されたプロパティのいずれも存在しない場合、null が書き込まれます。

• 構成: `property1|property2|...`

  列名に `|` 区切り文字が含まれている場合、この項目に対して複数のプロパティが選択されます。

• スクリプト構成:

  "multi":{    
      "multi":true
  }

  説明

  このパラメーターはコードレス UI では構成できません。

• 例:

  ## Reader: Elasticsearch の生データ
  "hits": [
      {
          "_index": "mutiltest_1",
          "_type": "_doc",
          "_id": "v3ShOoMB4GR_1DmrZN22",
          "_score": 1.0,
          "_source": {
              "feature1": "feature1",
              "feature2": [
                  1,
                  2,
                  3
              ],
              "feature3": {
                  "child": "feature3"
              }
          }
      }]
  
  ## Data Integration Elasticsearch reader プラグインの構成
  "parameter": {
    "column":[
          "feature1|feature2|feature3"
    ],
    "multi":{
          "multi":true
      }
  }
  
  ## Writer の結果: 1 行 1 列のデータ
  "feature1"

シナリオ 6: 複数のプロパティを選択的に同期する

• 背景: 複数のプロパティを選択的に処理できます。値を持つ最初のプロパティが返されます。指定されたプロパティのいずれも存在しない場合、null が書き込まれます。

• 構成: `property1|property2|...`

  列名に `|` 区切り文字が含まれている場合、この項目に対して複数のプロパティが選択されます。

• スクリプト構成:

  "multi":{
      "multi":true
  }

  説明

  このパラメーターはコードレス UI では構成できません。

• 例:

  ## Reader: Elasticsearch の生データ
  "hits": [
      {
          "_index": "mutiltest_1",
          "_type": "_doc",
          "_id": "v3ShOoMB4GR_1DmrZN22",
          "_score": 1.0,
          "_source": {
              "feature1": "feature1",
              "feature2": [
                  1,
                  2,
                  3
              ],
              "feature3": {
                  "child": "feature3"
              }
          }
      }]
  ## Data Integration Elasticsearch reader プラグインの構成
  "parameter": {
    "column":[
          "feature1,feature2,feature3"
    ],
    "multi":{
          "multi":true
      }
  }
  
  ## Writer の結果: 1 行 1 列のデータ
  "feature1,[1,2,3],{"child":"feature3"}"