DataWorks:Origen de datos de Elasticsearch

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

Tarea de sincronización sin conexión (tabla única)

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

• Para ver la referencia completa de parámetros y un script de ejemplo, consulte Appendix 1: Script demo and parameter descriptions.

Tarea de escritura en tiempo real (tabla única)

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

Sincronización en tiempo real de toda la base de datos

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

Configurar una tarea de sincronización por lotes mediante el editor de código

Si desea configurar una tarea de sincronización por lotes mediante el editor de código, defina los parámetros correspondientes en el script respetando los requisitos unificados de formato. Para obtener más información, consulte Script mode configuration. A continuación se describen los parámetros que debe configurar para los orígenes de datos al utilizar el editor de código.

Script de ejemplo de 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 del script de Reader

Script de ejemplo de 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 del script de 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"
      }

Escenario 1: Extracción completa de datos

• Contexto: extrae un documento completo de Elasticsearch en un solo campo.

• Ejemplo de configuración:

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

Escenario 2: Sincronizar propiedades de campos anidados u object

• Contexto: usa una ruta para acceder a las propiedades de un campo object o nested.

• Sintaxis:

  • property

  • property.sub-property

  • property[0].sub-property

• Configuración en el editor de código:

  "multi":{
      "multi":true
  }

  Nota

  La interfaz sin código no es compatible con esta función.

• Ejemplo de configuración:

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

  • Si un nodo antecesor de la ruta es una matriz, el resultado es null. Por ejemplo, recuperar level1.level2.level3 devuelve null porque level2 es una matriz. Para acceder a la propiedad, debe especificar un índice, como level1.level2[0].level3 o level1.level2[1].level3. La sintaxis de comodín level1.level2[*].level3 no es compatible.

  • Las claves que contienen un punto (.) no son compatibles. Por ejemplo, si los datos son {"level1.level2":{"level3":"testlevel3_1"}}, el resultado es null.

Escenario 3: Dividir una propiedad de matriz en varias filas

• Contexto: para relaciones de uno a varios, puede desplegar una matriz al dividir cada elemento en una fila independiente.

• Sintaxis: property[*].sub-property

• Ejemplo de efecto: los datos de origen como { "splitKey": [1, 2, 3, 4, 5] } se transforman en cinco filas, donde cada fila contiene un elemento de la matriz.

• Configuración en el editor de código:

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

  Nota

  • En la interfaz sin código, establezca el parámetro split multi-row array column name para lograr el mismo efecto, ya que genera automáticamente la configuración de script correspondiente.

  • El valor de la propiedad especificada por key debe ser una lista. De lo contrario, se produce un error.

• Ejemplo de configuración:

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

Escenario 4: Deduplicar y combinar propiedades de matriz

• Contexto: deduplica los elementos de la matriz y los combina en una cadena separada por comas. También funciona con subpropiedades como name1.name2. La deduplicación se basa en el resultado de toString() de cada elemento.

• Sintaxis: property[].

  Incluir [] en el nombre de la columna activa esta operación de deduplicación y combinación.

• Configuración en el editor de código:

  "multi":{
      "multi":true
  }

  Nota

  La interfaz sin código no es compatible con esta función.

• Ejemplo de configuración:

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

Escenario 5: Sincronizar selectivamente varias propiedades

• Contexto: devuelve el primer valor de propiedad distinto de null de una lista especificada. Si todas las propiedades son null o están ausentes, se devuelve null.

• Sintaxis: property1|property2|...

  Incluir el carácter de tubería (|) en el nombre de la columna activa esta selección de múltiples propiedades.

• Configuración en el editor de código:

  "multi":{    
      "multi":true
  }

  Nota

  La interfaz sin código no es compatible con esta función.

• Ejemplo de configuración:

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

Escenario 6: Combinar y sincronizar varias propiedades

• Contexto: concatena varios valores de propiedad en una sola cadena separada por comas.

• Sintaxis: property1,property2,...

  Incluir una coma (,) en el nombre de la columna activa esta operación de combinación de múltiples propiedades.

• Configuración en el editor de código:

  "multi":{
      "multi":true
  }

  Nota

  La interfaz sin código no es compatible con esta función.

• Ejemplo de configuración:

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