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 のクエリ本文と同じです。
                "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" // 最大レート（Mbps 単位。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 または nested フィールドのプロパティにパスを使用してアクセスします。

• 構文：

  • 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 結果：1 行、4 カラム
  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 を取得すると、level2 が配列であるため 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] } などのソースデータは、各配列要素を 1 行ずつ含む 5 行に変換されます。

• コードエディタ構成：

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

  説明

  • コードレス UI では、複数行配列カラム名の分割 パラメーターを設定することで、同じ効果を得られ、対応するスクリプト構成が自動生成されます。

  • key で指定されたプロパティの値はリストである必要があります。そうでない場合、エラーが発生します。

• 構成例：

  ## 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 でないプロパティ値を返します。すべてのプロパティが null または存在しない場合は、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：複数プロパティのマージおよび同期

• 背景情報：複数のプロパティ値を連結して、単一のカンマ区切り文字列にします。

• 構文：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\"}"