DataWorks:Fonte de dados do Elasticsearch

Tarefa de sincronização offline (tabela única)

• Consulte Configure a task in the codeless UI ou Configure a task in the code editor.

• Para a referência completa de parâmetros e uma demonstração de script, consulte Appendix 1: Script demo and parameter descriptions.

Tarefa de gravação em tempo real (tabela única)

Consulte Configure a real-time synchronization task in DataStudio (Legacy).

Sincronização em tempo real de banco de dados completo

Consulte Configure a real-time full-database synchronization task.

Configurar uma tarefa de sincronização em lote usando o editor de código

Para configurar uma tarefa de sincronização em lote no editor de código, defina os parâmetros relacionados no script de acordo com os requisitos de formato unificado. Para mais informações, consulte Script mode configuration. As informações a seguir descrevem os parâmetros obrigatórios para as fontes de dados ao configurar uma tarefa de sincronização em lote no editor de código.

Demonstração de script do Reader

{
    "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.
}

Parâmetros do script do Reader

Demonstração de script do Writer

{
    "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"
}

Parâmetros do script do Writer

- 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"
      }

Cenário 1: Extração completa de dados

• Contexto: Extrai um documento inteiro do Elasticsearch para um único campo.

• Configuração de exemplo:

  
  ## 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"]}

Cenário 2: Sincronizar propriedades de campos aninhados ou de objetos

• Contexto: Usa um caminho para acessar propriedades de um campo object ou nested.

• Sintaxe:

  • propriedade

  • propriedade.subpropriedade

  • propriedade[0].subpropriedade

• Configuração do editor de código:

  "multi":{
      "multi":true
  }

  Nota

  Este recurso não é suportado na interface sem código.

• Configuração de exemplo:

  ## 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

  Nota

  • Se um nó ancestral no caminho for um array, o resultado será null. Por exemplo, recuperar level1.level2.level3 retorna null porque level2 é um array. Para acessar a propriedade, especifique um índice, como level1.level2[0].level3 ou level1.level2[1].level3. A sintaxe de curinga level1.level2[*].level3 não é suportada.

  • Chaves que contêm um ponto (.) não são suportadas. Por exemplo, se os dados forem {"level1.level2":{"level3":"testlevel3_1"}}, o resultado será null.

Cenário 3: Dividir uma propriedade de array em várias linhas

• Contexto: Para relacionamentos de um para muitos, expanda um array dividindo cada elemento em uma linha separada.

• Sintaxe: property[*].sub-property

• Efeito de exemplo: Dados de origem como { "splitKey": [1, 2, 3, 4, 5] } são transformados em cinco linhas, com cada linha contendo um elemento do array.

• Configuração do editor de código:

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

  Nota

  • Na interface sem código, defina o parâmetro split multi-row array column name para obter o mesmo efeito gerando automaticamente a configuração de script correspondente.

  • O valor da propriedade especificada por key deve ser uma lista. Caso contrário, ocorrerá um erro.

• Configuração de exemplo:

  ## 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

Cenário 4: Deduplicar e mesclar propriedades de array

• Contexto: Deduplica elementos de array e os mescla em uma string separada por vírgulas. Isso também funciona com subpropriedades como name1.name2. A deduplicação é baseada no resultado toString() de cada elemento.

• Sintaxe: property[].

  Incluir [] no nome da coluna aciona essa operação de deduplicação e mesclagem.

• Configuração do editor de código:

  "multi":{
      "multi":true
  }

  Nota

  Este recurso não é suportado na interface sem código.

• Configuração de exemplo:

  ## 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"

Cenário 5: Sincronizar seletivamente várias propriedades

• Contexto: Retorna o primeiro valor de propriedade não nulo de uma lista especificada. Se todas as propriedades forem nulas ou estiverem ausentes, null será retornado.

• Sintaxe: property1|property2|...

  Incluir o caractere pipe (|) no nome da coluna aciona essa seleção de múltiplas propriedades.

• Configuração do editor de código:

  "multi":{    
      "multi":true
  }

  Nota

  Este recurso não é suportado na interface sem código.

• Configuração de exemplo:

  ## 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"

Cenário 6: Mesclar e sincronizar várias propriedades

• Contexto: Concatena vários valores de propriedade em uma única string separada por vírgulas.

• Sintaxe: property1,property2,...

  Incluir uma vírgula (,) no nome da coluna aciona essa operação de mesclagem de múltiplas propriedades.

• Configuração do editor de código:

  "multi":{
      "multi":true
  }

  Nota

  Este recurso não é suportado na interface sem código.

• Configuração de exemplo:

  ## 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\"}"