DataWorks:Source de données 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.
}

Tâche de synchronisation hors ligne (table unique)

• Consultez la section Configurer une tâche dans l'interface sans code ou Configurer une tâche dans l'éditeur de code.

• Pour une référence complète des paramètres et un exemple de script, consultez la section Annexe 1 : Exemple de script et descriptions des paramètres.

Tâche d'écriture en temps réel (table unique)

Consultez la section Configurer une tâche de synchronisation en temps réel dans DataStudio (hérité).

Synchronisation en temps réel de base de données complète

Consultez la section Configurer une tâche de synchronisation en temps réel de base de données complète.

Configurer une tâche de synchronisation par lots à l'aide de l'éditeur de code

Si vous souhaitez configurer une tâche de synchronisation par lots à l'aide de l'éditeur de code, vous devez configurer les paramètres associés dans le script selon les exigences de format de script unifié. Pour plus d'informations, consultez la section Configuration en mode script. Les informations suivantes décrivent les paramètres que vous devez configurer pour les sources de données lors de la configuration d'une tâche de synchronisation par lots à l'aide de l'éditeur de code.

Exemple de script pour le lecteur

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

Paramètres du script du lecteur

Exemple de script pour l'outil d'écriture

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

Paramètres du script de l'outil d'écriture

- id  // Le type id correspond à _id dans Elasticsearch et peut être considéré comme une clé primaire unique. Lors de l'écriture des données, les documents ayant le même id s'écrasent mutuellement et ne sont pas indexés.
- 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"
      }

Scénario 1 : Extraction complète des données

• Contexte : Extrait l'intégralité d'un document Elasticsearch dans un seul champ.

• Exemple de configuration :

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

Scénario 2 : Synchroniser les propriétés de champs imbriqués ou objet

• Contexte : Utilise un chemin d'accès pour accéder aux propriétés d'un champ object ou nested.

• Syntaxe :

  • property

  • property.sub-property

  • property[0].sub-property

• Configuration de l'éditeur de code :

  "multi":{
      "multi":true
  }

  Remarque

  Cette fonctionnalité n'est pas prise en charge dans l'interface sans code.

• Exemple de configuration :

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

  Remarque

  • Si un nœud ancêtre dans le chemin est un tableau, le résultat est null. Par exemple, la récupération de level1.level2.level3 renvoie null car level2 est un tableau. Pour accéder à la propriété, vous devez spécifier un index, tel que level1.level2[0].level3 ou level1.level2[1].level3. La syntaxe wildcard level1.level2[*].level3 n'est pas prise en charge.

  • Les clés contenant un point (.) ne sont pas prises en charge. Par exemple, si les données sont {"level1.level2":{"level3":"testlevel3_1"}}, le résultat est null.

Scénario 3 : Diviser une propriété de type tableau en plusieurs lignes

• Contexte : Pour les relations un-à-plusieurs, vous pouvez dérouler un tableau en divisant chaque élément en une ligne distincte.

• Syntaxe : property[*].sub-property

• Effet exemple : Les données source telles que { "splitKey": [1, 2, 3, 4, 5] } sont transformées en cinq lignes, chaque ligne contenant un élément du tableau.

• Configuration de l'éditeur de code :

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

  Remarque

  • Dans l'interface sans code, vous pouvez définir le paramètre split multi-row array column name pour obtenir le même effet en générant automatiquement la configuration de script correspondante.

  • La valeur de la propriété spécifiée par key doit être une liste. Sinon, une erreur se produit.

• Exemple de configuration :

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

Scénario 4 : Dédupliquer et fusionner les propriétés de type tableau

• Contexte : Déduplique les éléments du tableau et les fusionne en une chaîne séparée par des virgules. Cela fonctionne également avec les sous-propriétés telles que name1.name2. La déduplication est basée sur le résultat toString() de chaque élément.

• Syntaxe : property[].

  L'inclusion de [] dans le nom de la colonne déclenche cette opération de déduplication et de fusion.

• Configuration de l'éditeur de code :

  "multi":{
      "multi":true
  }

  Remarque

  Cette fonctionnalité n'est pas prise en charge dans l'interface sans code.

• Exemple de configuration :

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

Scénario 5 : Synchroniser sélectivement plusieurs propriétés

• Contexte : Renvoie la première valeur de propriété non nulle d'une liste spécifiée. Si toutes les propriétés sont nulles ou absentes, null est renvoyé.

• Syntaxe : property1|property2|...

  L'inclusion du caractère pipe (|) dans le nom de la colonne déclenche cette sélection multi-propriétés.

• Configuration de l'éditeur de code :

  "multi":{    
      "multi":true
  }

  Remarque

  Cette fonctionnalité n'est pas prise en charge dans l'interface sans code.

• Exemple de configuration :

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

Scénario 6 : Fusionner et synchroniser plusieurs propriétés

• Contexte : Concatène plusieurs valeurs de propriété en une seule chaîne séparée par des virgules.

• Syntaxe : property1,property2,...

  L'inclusion d'une virgule (,) dans le nom de la colonne déclenche cette opération de fusion multi-propriétés.

• Configuration de l'éditeur de code :

  "multi":{
      "multi":true
  }

  Remarque

  Cette fonctionnalité n'est pas prise en charge dans l'interface sans code.

• Exemple de configuration :

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