DataWorks:Sumber data Elasticsearch

Tugas sinkronisasi offline (tabel tunggal)

• Lihat Konfigurasikan tugas di Antarmuka tanpa kode atau Konfigurasikan tugas di editor kode.

• Untuk referensi parameter lengkap dan demo skrip, lihat Lampiran 1: Demo skrip dan deskripsi parameter.

Tugas tulis real-time (tabel tunggal)

Lihat Konfigurasikan tugas sinkronisasi real-time di DataStudio (Legacy).

Sinkronisasi real-time seluruh database

Lihat Konfigurasikan tugas sinkronisasi real-time seluruh database.

Konfigurasikan tugas sinkronisasi batch menggunakan editor kode

Jika ingin mengonfigurasi tugas sinkronisasi batch menggunakan editor kode, Anda harus mengonfigurasi parameter terkait dalam skrip berdasarkan persyaratan format skrip terpadu. Untuk informasi lebih lanjut, lihat Konfigurasi mode skrip. Informasi berikut menjelaskan parameter yang harus dikonfigurasi untuk sumber data saat mengatur tugas sinkronisasi batch menggunakan editor kode.

Demo skrip Reader

{
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    },
    "setting":{
        "errorLimit":{
            "record":"0" // Jumlah catatan error yang diizinkan.
        },
        "jvmOption":"",
        "speed":{
            "concurrent":3,
            "throttle":false
        }
    },
    "steps":[
        {
            "category":"reader",
            "name":"Reader",
            "parameter":{
                "column":[ // Kolom yang akan dibaca.
                    "id",
                    "name"
                ],
                "endpoint":"http://es-cn-xxx.elasticsearch.aliyuncs.com:9200", // Titik akhir.
                "index":"aliyun_es_xx",  // Index.
                "password":"*******",  // Kata sandi.
                "multiThread":true,
                "scroll":"5m",  // Durasi konteks scroll.
                "pageSize":5000,
                "connTimeOut":600000,
                "readTimeOut":600000,
                "retryCount":30,
                "retrySleepTime":"10000",
                "search":{
                            "range":{
                                "gmt_modified":{
                                    "gte":0
                                }
                            }
                        },  // Parameter kueri. Isinya identik dengan badan kueri API _search Elasticsearch.
                "type":"doc",
                "username":"aliyun_di"  // Nama pengguna.
            },
            "stepType":"elasticsearch"
        },
        {
            "category":"writer",
            "name":"Writer",
            "parameter":{ },
            "stepType":"stream"
        }
    ],
    "type":"job",
    "version":"2.0" // Nomor versi.
}

Parameter skrip Reader

Demo skrip Writer

{
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    },
    "setting": {
        "errorLimit": {
            "record": "0"
        },
        "speed": {
            "throttle":true, // Jika throttle bernilai false, parameter mbps diabaikan dan pembatasan kecepatan dinonaktifkan. Jika throttle bernilai true, pembatasan kecepatan diaktifkan.
            "concurrent":1, // Konkurensi pekerjaan.
            "mbps":"12" // Laju maksimum dalam 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"
}

Parameter skrip Writer

- id  // Tipe id dipetakan ke _id di Elasticsearch dan dapat dianggap sebagai kunci primer unik. Saat menulis data, dokumen dengan id yang sama akan saling menimpa dan tidak diindeks.
- 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"
  }

Skenario 1: Tarik seluruh data

• Latar belakang: Menarik seluruh dokumen Elasticsearch ke dalam satu field.

• Contoh konfigurasi:

  
  ## reader: Data asli di Elasticsearch
  "hits": [
      {
          "_index": "mutiltest_1",
          "_type": "_doc",
          "_id": "IXgdO4MB4GR_1DmrjTXP",
          "_score": 1.0,
          "_source": {
              "feature1": "value1",
              "feature2": "value2",
              "feature3": "value3"
          }
      }]
  
  ## Konfigurasi reader Elasticsearch Integrasi Data
  "parameter": {
    "column": [
        "content"
    ],
    "full":true
  }
  
  ## Hasil writer: Menulis satu baris dan satu kolom ke tujuan.
  {"_index":"mutiltest_1","_type":"_doc","_id":"IXgdO4MB4GR_1DmrjTXP","_source":{"feature1":"value1","feature2":"value2","feature3":"value3"},"sort":["IXgdO4MB4GR_1DmrjTXP"]}

Skenario 2: Sinkronkan properti field nested atau object

• Latar belakang: Menggunakan path untuk mengakses properti field object atau nested.

• Sintaks:

  • property

  • property.sub-property

  • property[0].sub-property

• Konfigurasi editor kode:

  "multi":{
      "multi":true
  }

  Catatan

  Fitur ini tidak didukung di Antarmuka tanpa kode.

• Contoh konfigurasi:

  ## reader: Data asli di Elasticsearch
  "hits": [
      {
          "_index": "mutiltest_1",
          "_type": "_doc",
          "_id": "7XAOOoMB4GR_1Dmrrust",
          "_score": 1.0,
          "_source": {
              "level1": {
                  "level2": [
                      {
                          "level3": "testlevel3_1"
                      },
                      {
                          "level3": "testlevel3_2"
                      }
                  ]
              }
          }
      }
  ]
  ## Konfigurasi reader Elasticsearch Integrasi Data
  "parameter": {
    "column": [
        "level1",
        "level1.level2",
        "level1.level2[0]",
        "level1.level2.level3"
    ],
    "multi":{
          "multi":true
      }
  }
  
  ## Hasil writer: 1 baris, 4 kolom
  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

  Catatan

  • Jika node leluhur dalam path berupa array, hasilnya adalah null. Misalnya, mengambil level1.level2.level3 menghasilkan null karena level2 adalah array. Untuk mengakses properti tersebut, Anda harus menentukan indeks, seperti level1.level2[0].level3 atau level1.level2[1].level3. Sintaks wildcard level1.level2[*].level3 tidak didukung.

  • Kunci yang mengandung titik (.) tidak didukung. Misalnya, jika datanya adalah {"level1.level2":{"level3":"testlevel3_1"}}, hasilnya adalah null.

Skenario 3: Pisahkan properti array menjadi beberapa baris

• Latar belakang: Untuk relasi satu-ke-banyak, Anda dapat membuka array dengan membagi setiap elemen menjadi baris terpisah.

• Sintaks: property[*].sub-property

• Contoh efek: Data sumber seperti { "splitKey": [1, 2, 3, 4, 5] } diubah menjadi lima baris, dengan setiap baris berisi satu elemen dari array.

• Konfigurasi editor kode:

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

  Catatan

  • Di Antarmuka tanpa kode, Anda dapat mengatur parameter split multi-row array column name untuk mencapai efek yang sama dengan menghasilkan konfigurasi skrip yang sesuai secara otomatis.

  • Nilai properti yang ditentukan oleh key harus berupa daftar. Jika tidak, terjadi error.

• Contoh konfigurasi:

  ## reader: Data asli di 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"
                  }
              ]
          }
      }
  ]
  ## Konfigurasi reader Elasticsearch Integrasi Data
  {
     "column":[
        "headers[*].remoteip"
    ]
    "multi":{
        "multi":true,
        "key": "headers"
    }
  }
  
  ## Hasil writer: 4 baris
  192.0.2.1
  192.0.2.2
  192.0.2.3
  192.0.2.4

Skenario 4: Hapus duplikat dan gabungkan properti array

• Latar belakang: Menghapus duplikat elemen array dan menggabungkannya menjadi string yang dipisahkan koma. Fitur ini juga berlaku untuk sub-properti seperti name1.name2. Penghapusan duplikat didasarkan pada hasil toString() setiap elemen.

• Sintaks: property[].

  Menyertakan [] dalam nama kolom memicu operasi penghapusan duplikat dan penggabungan ini.

• Konfigurasi editor kode:

  "multi":{
      "multi":true
  }

  Catatan

  Fitur ini tidak didukung di Antarmuka tanpa kode.

• Contoh konfigurasi:

  ## reader: Data asli di Elasticsearch
  "hits": [
  {
      "_index": "mutiltest_1",
      "_type": "_doc",
      "_id": "4nbUOoMB4GR_1Dmryj8O",
      "_score": 1.0,
      "_source": {
          "feature1": [
              "value1",
              "value1",
              "value2",
              "value2",
              "value3"
          ]
      }
  }
  ]
  ## Konfigurasi reader Elasticsearch Integrasi Data
  "parameter": {
    "column":[
          "feature1[]"
    ],
    "multi":{
          "multi":true
      }
  }
  
  ## Hasil writer: 1 baris, 1 kolom
  "value1,value2,value3"

Skenario 5: Sinkronisasi selektif beberapa properti

• Latar belakang: Mengembalikan nilai properti pertama yang tidak null dari daftar yang ditentukan. Jika semua properti bernilai null atau tidak ada, null dikembalikan.

• Sintaks: property1|property2|...

  Menyertakan karakter pipa (|) dalam nama kolom memicu pemilihan multi-properti ini.

• Konfigurasi editor kode:

  "multi":{    
      "multi":true
  }

  Catatan

  Fitur ini tidak didukung di Antarmuka tanpa kode.

• Contoh konfigurasi:

  ## reader: Data asli di Elasticsearch
  "hits": [
      {
          "_index": "mutiltest_1",
          "_type": "_doc",
          "_id": "v3ShOoMB4GR_1DmrZN22",
          "_score": 1.0,
          "_source": {
              "feature1": "feature1",
              "feature2": [
                  1,
                  2,
                  3
              ],
              "feature3": {
                  "child": "feature3"
              }
          }
      }]
  
  ## Konfigurasi reader Elasticsearch Integrasi Data
  "parameter": {
    "column":[
          "feature1|feature2|feature3"
    ],
    "multi":{
          "multi":true
      }
  }
  
  ## Hasil writer: 1 baris, 1 kolom
  "feature1"

Skenario 6: Gabungkan dan sinkronkan beberapa properti

• Latar belakang: Menggabungkan nilai beberapa properti menjadi satu string yang dipisahkan koma.

• Sintaks: property1,property2,...

  Menyertakan koma (,) dalam nama kolom memicu operasi penggabungan multi-properti ini.

• Konfigurasi editor kode:

  "multi":{
      "multi":true
  }

  Catatan

  Fitur ini tidak didukung di Antarmuka tanpa kode.

• Contoh konfigurasi:

  ## reader: Data asli di Elasticsearch
  "hits": [
      {
          "_index": "mutiltest_1",
          "_type": "_doc",
          "_id": "v3ShOoMB4GR_1DmrZN22",
          "_score": 1.0,
          "_source": {
              "feature1": "feature1",
              "feature2": [
                  1,
                  2,
                  3
              ],
              "feature3": {
                  "child": "feature3"
              }
          }
      }]
  ## Konfigurasi reader Elasticsearch Integrasi Data
  "parameter": {
    "column":[
          "feature1,feature2,feature3"
    ],
    "multi":{
          "multi":true
      }
  }
  
  ## Hasil writer: 1 baris, 1 kolom
  "feature1,[1,2,3],{\"child\":\"feature3\"}"