DataWorks:Elasticsearch veri kaynağı

{
 "order":{
  "hops":[
   {
    "from":"Reader",
    "to":"Writer"
   }
  ]
 },
 "setting":{
  "errorLimit":{
   "record":"0" // The error limit. The job stops if the error count exceeds this value.
  },
  "jvmOption":"",
  "speed":{
   "concurrent":3, // The number of concurrent threads.
   "throttle":true,
   "mbps":"12"    // The maximum speed in MB/s.
  }
 },
 "steps":[
  {
   "category":"reader",
   "name":"Reader",
   "parameter":{
    "column":[ // The columns to read.
     "id",
     "name"
    ],
    "endpoint":"", // The service endpoint.
    "index":"",  // The source index.
    "password":"",  // The password for authentication.
    "scroll":"",  // The scroll context duration.
    "search":"",  // The query body, same as the `query` object in an Elasticsearch `_search` request.
    "type":"default",
    "username":""  // The username for authentication.
   },
   "stepType":"elasticsearch"
  },
  {
   "stepType": "elasticsearch",
            "parameter": {
                "column": [ // The columns to write.
                    {
                        "name": "id",
                        "type": "integer"
                    },
                    {
                        "name": "name",
                        "type": "text"
                    }
                ],
                "index": "test",   // The destination index.
                 "indexType": "",   // The index type. Leave this empty for Elasticsearch v7.x or later.
                "actionType": "index",  // The write mode.
                "cleanup": false,         // Specifies whether to recreate the index before writing.
                "datasource": "test",   // The data source name.
                "primaryKeyInfo": {     // The primary key configuration.
                    "fieldDelimiterOrigin": ",",
                    "column": [
                        "id"
                    ],
                    "type": "specific",
                    "fieldDelimiter": ","
                },
                "dynamic": false,  // Specifies whether to enable dynamic mapping.
                "batchSize": 1024   // The number of documents to write per batch.
            },
            "name": "Writer",
            "category": "writer"
  }
 ],
 "type":"job",
 "version":"2.0" // The version number.
}

Çevrimdışı senkronizasyon görevi (tek tablo)

• Bkz. Kodsuz kullanıcı arayüzünde görev yapılandırma veya Kod düzenleyicide görev yapılandırma.

• Tam parametre başvurusu ve bir komut dosyası demosu için bkz. Ek 1: Komut dosyası demosu ve parametre açıklamaları.

Gerçek zamanlı yazma görevi (tek tablo)

Bkz. DataStudio'da (Eski) gerçek zamanlı senkronizasyon görevi yapılandırma.

Tam veritabanı gerçek zamanlı senkronizasyon

Bkz. Gerçek zamanlı tam veritabanı senkronizasyon görevi yapılandırma.

Kod düzenleyiciyi kullanarak toplu senkronizasyon görevi yapılandırma

Kod düzenleyiciyi kullanarak toplu senkronizasyon görevi yapılandırmak istiyorsanız, komut dosyasındaki ilgili parametreleri birleşik komut dosyası biçimi gereksinimlerine göre yapılandırmanız gerekir. Daha fazla bilgi için bkz. Komut dosyası modu yapılandırması. Aşağıdaki bilgiler, kod düzenleyiciyi kullanarak toplu senkronizasyon görevi yapılandırırken veri kaynakları için yapılandırmanız gereken parametreleri açıklar.

Reader komut dosyası demosu

{
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    },
    "setting":{
        "errorLimit":{
            "record":"0" // The number of error records allowed.
        },
        "jvmOption":"",
        "speed":{
            "concurrent":3,
            "throttle":false
        }
    },
    "steps":[
        {
            "category":"reader",
            "name":"Reader",
            "parameter":{
                "column":[ // The columns to read.
                    "id",
                    "name"
                ],
                "endpoint":"http://es-cn-xxx.elasticsearch.aliyuncs.com:9200", // The endpoint.
                "index":"aliyun_es_xx",  // The index.
                "password":"*******",  // The password.
                "multiThread":true,
                "scroll":"5m",  // The scroll context duration.
                "pageSize":5000,
                "connTimeOut":600000,
                "readTimeOut":600000,
                "retryCount":30,
                "retrySleepTime":"10000",
                "search":{
                            "range":{
                                "gmt_modified":{
                                    "gte":0
                                }
                            }
                        },  // Query parameter. The content is identical to the query body of the Elasticsearch _search API.
                "type":"doc",
                "username":"aliyun_di"  // The username.
            },
            "stepType":"elasticsearch"
        },
        {
            "category":"writer",
            "name":"Writer",
            "parameter":{ },
            "stepType":"stream"
        }
    ],
    "type":"job",
    "version":"2.0" // The version number.
}

Reader komut dosyası parametreleri

Writer komut dosyası demosu

{
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    },
    "setting": {
        "errorLimit": {
            "record": "0"
        },
        "speed": {
            "throttle":true, // If throttle is false, the mbps parameter is ignored and throttling is disabled. If throttle is true, throttling is enabled.
            "concurrent":1, // The job concurrency.
            "mbps":"12" // The maximum rate in 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 komut dosyası parametreleri

- id  // The id type maps to _id in Elasticsearch and can be considered a unique primary key. When writing data, documents with the same id overwrite each other and are not indexed.
- 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"
      }

Senaryo 1: Tam veri çekme

• Arka plan: Tüm bir Elasticsearch belgesini tek bir alana çeker.

• Örnek yapılandırma:

  
  ## reader: Original data in Elasticsearch
  "hits": [
      {
          "_index": "mutiltest_1",
          "_type": "_doc",
          "_id": "IXgdO4MB4GR_1DmrjTXP",
          "_score": 1.0,
          "_source": {
              "feature1": "value1",
              "feature2": "value2",
              "feature3": "value3"
          }
      }]
  
  ## Data Integration Elasticsearch reader configuration
  "parameter": {
    "column": [
        "content"
    ],
    "full":true
  }
  
  ## writer result: Writes one row and one column to the destination.
  {"_index":"mutiltest_1","_type":"_doc","_id":"IXgdO4MB4GR_1DmrjTXP","_source":{"feature1":"value1","feature2":"value2","feature3":"value3"},"sort":["IXgdO4MB4GR_1DmrjTXP"]}

Senaryo 2: İç içe geçmiş veya nesne alanı özelliklerini senkronize etme

• Arka plan: Bir object veya nested alanının özelliklerine erişmek için bir yol kullanır.

• Sözdizimi:

  • property

  • property.sub-property

  • property[0].sub-property

• Kod düzenleyici yapılandırması:

  "multi":{
      "multi":true
  }

  Not

  Bu özellik kodsuz kullanıcı arayüzünde desteklenmez.

• Örnek yapılandırma:

  ## reader: Original data in 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 configuration
  "parameter": {
    "column": [
        "level1",
        "level1.level2",
        "level1.level2[0]",
        "level1.level2.level3"
    ],
    "multi":{
          "multi":true
      }
  }
  
  ## writer result: 1 row, 4 columns
  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

  Not

  • Yoldaki bir üst düğüm bir dizi ise sonuç null olur. Örneğin, level1.level2.level3 alma işlemi level2 bir dizi olduğu için null döndürür. Özelliğe erişmek için level1.level2[0].level3 veya level1.level2[1].level3 gibi bir dizin belirtmeniz gerekir. level1.level2[*].level3 joker karakter sözdizimi desteklenmez.

  • Nokta (.) içeren anahtarlar desteklenmez. Örneğin, veri {"level1.level2":{"level3":"testlevel3_1"}} ise sonuç null olur.

Senaryo 3: Bir dizi özelliğini birden fazla satıra ayırma

• Arka plan: Bire çok ilişkiler için her öğeyi ayrı bir satıra bölerek bir diziyi genişletebilirsiniz.

• Sözdizimi: property[*].sub-property

• Örnek etki: { "splitKey": [1, 2, 3, 4, 5] } gibi kaynak veriler, diziden her biri bir öğe içeren beş satıra dönüştürülür.

• Kod düzenleyici yapılandırması:

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

  Not

  • Kodsuz kullanıcı arayüzünde aynı etkiyi elde etmek için split multi-row array column name parametresini ayarlayarak ilgili komut dosyası yapılandırmasını otomatik olarak oluşturabilirsiniz.

  • key ile belirtilen özelliğin değeri bir liste olmalıdır. Aksi takdirde hata oluşur.

• Örnek yapılandırma:

  ## reader: Original data in 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 configuration
  {
     "column":[
        "headers[*].remoteip"
    ]
    "multi":{
        "multi":true,
        "key": "headers"
    }
  }
  
  ## writer result: 4 rows
  192.0.2.1
  192.0.2.2
  192.0.2.3
  192.0.2.4

Senaryo 4: Dizi özelliklerini yinelemesiz hale getirme ve birleştirme

• Arka plan: Dizi öğelerini yinelenmeyen hale getirir ve virgülle ayrılmış bir dizede birleştirir. Bu işlem name1.name2 gibi alt özelliklerle de çalışır. Yinelenmeme işlemi her öğenin toString() sonucuna dayanır.

• Sözdizimi: property[].

  Sütun adında [] bulunması bu yinelenmeme ve birleştirme işlemini tetikler.

• Kod düzenleyici yapılandırması:

  "multi":{
      "multi":true
  }

  Not

  Bu özellik kodsuz kullanıcı arayüzünde desteklenmez.

• Örnek yapılandırma:

  ## reader: Original data in Elasticsearch
  "hits": [
  {
      "_index": "mutiltest_1",
      "_type": "_doc",
      "_id": "4nbUOoMB4GR_1Dmryj8O",
      "_score": 1.0,
      "_source": {
          "feature1": [
              "value1",
              "value1",
              "value2",
              "value2",
              "value3"
          ]
      }
  }
  ]
  ## Data Integration Elasticsearch reader configuration
  "parameter": {
    "column":[
          "feature1[]"
    ],
    "multi":{
          "multi":true
      }
  }
  
  ## writer result: 1 row, 1 column
  "value1,value2,value3"

Senaryo 5: Birden fazla özelliği seçici olarak senkronize etme

• Arka plan: Belirtilen listeden ilk boş olmayan özellik değerini döndürür. Tüm özellikler null veya yoksa null döndürülür.

• Sözdizimi: property1|property2|...

  Sütun adında dikey çizgi (|) karakterinin bulunması bu çoklu özellik seçimini tetikler.

• Kod düzenleyici yapılandırması:

  "multi":{    
      "multi":true
  }

  Not

  Bu özellik kodsuz kullanıcı arayüzünde desteklenmez.

• Örnek yapılandırma:

  ## reader: Original data in 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 configuration
  "parameter": {
    "column":[
          "feature1|feature2|feature3"
    ],
    "multi":{
          "multi":true
      }
  }
  
  ## writer result: 1 row, 1 column
  "feature1"

Senaryo 6: Birden fazla özelliği birleştirme ve senkronize etme

• Arka plan: Birden fazla özellik değerini tek bir virgülle ayrılmış dizede birleştirir.

• Sözdizimi: property1,property2,...

  Sütun adında virgül (,) bulunması bu çoklu özellik birleştirme işlemini tetikler.

• Kod düzenleyici yapılandırması:

  "multi":{
      "multi":true
  }

  Not

  Bu özellik kodsuz kullanıcı arayüzünde desteklenmez.

• Örnek yapılandırma:

  ## reader: Original data in 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 configuration
  "parameter": {
    "column":[
          "feature1,feature2,feature3"
    ],
    "multi":{
          "multi":true
      }
  }
  
  ## writer result: 1 row, 1 column
  "feature1,[1,2,3],{\"child\":\"feature3\"}"