Quais capacidades de sincronização de dados do Elasticsearch o DataWorks suporta - DataWorks: Central de Documentação do Alibaba Cloud

Versões suportadas e grupos de recursos

O DataWorks oferece suporte ao Alibaba Cloud Elasticsearch nas versões 5.x, 6.x, 7.x e 8.x. Fontes de dados Elasticsearch autogerenciadas não são suportadas.

O suporte a versões varia conforme o grupo de recursos:

Grupo de recursos	Versões suportadas
Grupo de recursos público	Elasticsearch 5.x
Grupo de recursos serverless (recomendado)	Elasticsearch 5.x, 6.x, 7.x, 8.x
Grupo de recursos exclusivo para Data Integration	Elasticsearch 5.x, 6.x, 7.x, 8.x

Para obter instruções de configuração, consulte Usar um grupo de recursos serverless e Usar grupos de recursos exclusivos para Data Integration.

Limitações

O Elasticsearch Reader busca informações de shard no servidor antes da sincronização. Todos os shards devem estar ativos durante o processo; shards inativos causam inconsistência de dados.
Para Elasticsearch 6.x ou superior, use um grupo de recursos serverless ou um grupo de recursos exclusivo para Data Integration. Grupos de recursos públicos não oferecem suporte ao Elasticsearch 6.x e versões posteriores.
Campos do tipo scaled_float não podem ser sincronizados.
Índices com $ref nos nomes dos campos não podem ser sincronizados.

Conceitos principais

O Elasticsearch mapeia para um banco de dados relacional da seguinte forma:

Elasticsearch	Banco de dados relacional
Elasticsearch (instância)	Banco de dados relacional (instância)
Índice	Bancos de dados
Tipos	Tabelas
Documentos	Linhas
Campos	Colunas

Tipos de campo suportados

Tipo	Leitura offline	Gravação offline	Gravação em tempo real
binary	Suportado	Suportado	Suportado
boolean	Suportado	Suportado	Suportado
keyword	Suportado	Suportado	Suportado
constant_keyword	Não suportado	Não suportado	Não suportado
wildcard	Não suportado	Não suportado	Não suportado
long	Suportado	Suportado	Suportado
integer	Suportado	Suportado	Suportado
short	Suportado	Suportado	Suportado
byte	Suportado	Suportado	Suportado
double	Suportado	Suportado	Suportado
float	Suportado	Suportado	Suportado
half_float	Não suportado	Não suportado	Não suportado
scaled_float	Não suportado	Não suportado	Não suportado
unsigned_long	Não suportado	Não suportado	Não suportado
date	Suportado	Suportado	Suportado
date_nanos	Não suportado	Não suportado	Não suportado
alias	Não suportado	Não suportado	Não suportado
object	Suportado	Suportado	Suportado
flattened	Não suportado	Não suportado	Não suportado
nested	Suportado	Suportado	Suportado
join	Não suportado	Não suportado	Não suportado
integer_range	Suportado	Suportado	Suportado
float_range	Suportado	Suportado	Suportado
long_range	Suportado	Suportado	Suportado
double_range	Suportado	Suportado	Suportado
date_range	Suportado	Suportado	Suportado
ip_range	Não suportado	Suportado	Suportado
ip	Suportado	Suportado	Suportado
version	Suportado	Suportado	Suportado
murmur3	Não suportado	Não suportado	Não suportado
aggregate_metric_double	Não suportado	Não suportado	Não suportado
histogram	Não suportado	Não suportado	Não suportado
text	Suportado	Suportado	Suportado
annotated-text	Não suportado	Não suportado	Não suportado
completion	Suportado	Não suportado	Não suportado
search_as_you_type	Não suportado	Não suportado	Não suportado
token_count	Suportado	Não suportado	Não suportado
dense_vector	Não suportado	Não suportado	Não suportado
rank_feature	Não suportado	Não suportado	Não suportado
rank_features	Não suportado	Não suportado	Não suportado
geo_point	Suportado	Suportado	Suportado
geo_shape	Suportado	Suportado	Suportado
point	Não suportado	Não suportado	Não suportado
shape	Não suportado	Não suportado	Não suportado
percolator	Não suportado	Não suportado	Não suportado
string	Suportado	Suportado	Suportado

Funcionamento

O Elasticsearch Reader utiliza a API de slice scroll _search para ler dados. O mecanismo de slice opera em conjunto com o sharding multithread nas tarefas do Data Integration, permitindo que múltiplas threads leiam de diferentes slices em paralelo. A conversão de tipos de dados ocorre com base na configuração de mapeamento do Elasticsearch.

Para mais informações sobre a API scroll, consulte a documentação oficial do Elasticsearch.

Recursos avançados

Além das leituras padrão no nível de campo, o Elasticsearch Reader oferece dois modos avançados de extração de dados:

Extração completa de documento: lê um documento inteiro do Elasticsearch como um único campo de string JSON. Consulte Cenário 1: Extração completa de dados.
Extração de semiestruturado para estruturado: achata JSON aninhado, divide arrays em linhas, remove duplicatas de valores de array, mescla várias propriedades ou seleciona a primeira propriedade não nula. Consulte Cenários 2–6 para detalhes de configuração.

Adicionar uma fonte de dados

Antes de configurar uma tarefa de sincronização, adicione a fonte de dados Elasticsearch no DataWorks. Siga as instruções em Gerenciamento de fontes de dados. As descrições dos parâmetros estão disponíveis diretamente no console do DataWorks ao adicionar a fonte de dados.

Desenvolver uma tarefa de sincronização de dados

Configurar uma tarefa de sincronização offline (tabela única)

Para um guia passo a passo, consulte Configurar uma tarefa na interface visual ou Configurar uma tarefa no editor de código.
Para a referência completa de parâmetros e exemplos de script, consulte Apêndice 1: Demonstrações de script e descrições de parâmetros.

Configurar uma tarefa de gravação em tempo real (tabela única)

Consulte Configurar uma tarefa de sincronização em tempo real no DataStudio.

Configurar uma tarefa de sincronização offline ou em tempo real de banco de dados completo

Consulte Configurar uma tarefa de sincronização de banco de dados completo em tempo real.

Configuração básica

O exemplo a seguir apresenta uma configuração mínima de Reader para Writer. Remova todos os comentários antes de executar.

{
  "order": {
    "hops": [
      {
        "from": "Reader",
        "to": "Writer"
      }
    ]
  },
  "setting": {
    "errorLimit": {
      "record": "0"         // Maximum number of error records allowed.
    },
    "jvmOption": "",
    "speed": {
      "concurrent": 3,      // Number of concurrent threads.
      "throttle": true,
      "mbps": "12"          // Throttle rate. 1 Mbps = 1 MB/s.
    }
  },
  "steps": [
    {
      "category": "reader",
      "name": "Reader",
      "parameter": {
        "column": [         // Columns to read.
          "id",
          "name"
        ],
        "endpoint": "",     // Elasticsearch endpoint.
        "index": "",        // Index name.
        "password": "",
        "scroll": "",       // Scroll context duration (e.g., "5m").
        "search": "",       // Query body — same format as Elasticsearch _search API.
        "type": "default",
        "username": ""
      },
      "stepType": "elasticsearch"
    },
    {
      "category": "writer",
      "name": "Writer",
      "parameter": {
        "column": [         // Columns to write, with type definitions.
          {
            "name": "id",
            "type": "integer"
          },
          {
            "name": "name",
            "type": "text"
          }
        ],
        "index": "test",            // Destination index.
        "indexType": "",            // Leave blank for Elasticsearch 7.x.
        "actionType": "index",      // Write mode: index or update.
        "cleanup": false,           // If true, deletes and recreates the index before writing.
        "datasource": "test",       // Data source name configured in DataWorks.
        "primaryKeyInfo": {
          "fieldDelimiterOrigin": ",",
          "column": ["id"],
          "type": "specific",
          "fieldDelimiter": ","
        },
        "dynamic": false,           // If false, mappings are generated from column config.
        "batchSize": 1024           // Documents written per batch.
      },
      "name": "Writer",
      "category": "writer",
      "stepType": "elasticsearch"
    }
  ],
  "type": "job",
  "version": "2.0"
}

Apêndice 1: Demonstrações de script e descrições de parâmetros

Demonstração de script do Reader

{
  "order": {
    "hops": [
      {
        "from": "Reader",
        "to": "Writer"
      }
    ]
  },
  "setting": {
    "errorLimit": {
      "record": "0"
    },
    "jvmOption": "",
    "speed": {
      "concurrent": 3,
      "throttle": false
    }
  },
  "steps": [
    {
      "category": "reader",
      "name": "Reader",
      "parameter": {
        "column": [
          "id",
          "name"
        ],
        "endpoint": "http://es-cn-xxx.elasticsearch.aliyuncs.com:9200",
        "index": "aliyun_es_xx",
        "password": "*******",
        "multiThread": true,
        "scroll": "5m",
        "pageSize": 5000,
        "connTimeOut": 600000,
        "readTimeOut": 600000,
        "retryCount": 30,
        "retrySleepTime": "10000",
        "search": {
          "range": {
            "gmt_modified": {
              "gte": 0
            }
          }
        },
        "type": "doc",
        "username": "aliyun_di"
      },
      "stepType": "elasticsearch"
    },
    {
      "category": "writer",
      "name": "Writer",
      "parameter": {},
      "stepType": "stream"
    }
  ],
  "type": "job",
  "version": "2.0"
}

Parâmetros do script do Reader

Parâmetro	Descrição	Obrigatório	Valor padrão
`datasource`	Nome da fonte de dados conforme configurada no DataWorks. Deve corresponder exatamente ao nome da fonte adicionada.	Sim	Nenhum
`index`	Nome do índice do Elasticsearch.	Sim	Nenhum
`type`	Tipo de índice do Elasticsearch.	Não	Nome do índice
`search`	Corpo da consulta do Elasticsearch. Utiliza o mesmo formato da API `_search`.	Sim	Nenhum
`pageSize`	Quantidade de registros a serem lidos por página.	Não	100
`scroll`	Duração do contexto de scroll. Controla por quanto tempo o contexto permanece ativo entre as leituras de página. Se definido com valor muito baixo, o scroll expira quando o tempo ocioso entre páginas é longo, causando perda de dados. Se definido com valor muito alto, consultas concorrentes podem exceder o limite `max_open_scroll_context` do servidor, gerando erros de consulta.	Sim	Nenhum
`strictMode`	Se `true` e ocorrer um erro `shard.failed`, a leitura é interrompida imediatamente para evitar a sincronização parcial de dados.	Não	true
`sort`	Campo utilizado para ordenar os resultados.	Não	Nenhum
`retryCount`	Número de tentativas após uma falha.	Não	300
`connTimeOut`	Tempo limite de conexão do cliente em milissegundos.	Não	600.000
`readTimeOut`	Tempo limite de leitura do cliente em milissegundos.	Não	600.000
`multiThread`	Indica se devem ser usadas múltiplas threads para requisições HTTP.	Não	true
`preemptiveAuth`	Indica se deve ser usada autenticação preemptiva para requisições HTTP.	Não	false
`retrySleepTime`	Intervalo entre novas tentativas em milissegundos.	Não	1.000
`discovery`	Modo de descoberta de nós. `true`: conecta-se a um nó aleatório do cluster e atualiza periodicamente a lista de servidores. `false`: envia todas as requisições para o endpoint configurado.	Não	false
`compression`	Indica se os corpos das requisições devem ser compactados com GZIP. Requer que `http.compression` esteja habilitado no nó do Elasticsearch.	Não	false
`dateFormat`	Obrigatório caso um campo `date` não possua `format` em seu mapeamento. Especifique todos os formatos de data usados no campo, por exemplo: `"yyyy-MM-dd		yyyy-MM-dd HH:mm:ss"`.	Não	Nenhum
`full`	Se `true`, lê o documento inteiro como um único campo de string JSON. Consulte Cenário 1: Extração completa de dados.	Não	Nenhum
`multi`	Habilita a extração avançada de múltiplas propriedades. Possui duas subpropriedades: `multi.key` e `multi.mult`. Consulte Recursos avançados para os cinco casos de uso suportados.	Não	Nenhum

Demonstração de script do Writer

Uma instância do Elasticsearch dentro de uma Virtual Private Cloud (VPC) não pode ser acessada por um grupo de recursos público devido ao isolamento de rede. Use um grupo de recursos serverless ou um grupo de recursos exclusivo para Data Integration para acessar instâncias em uma VPC.

{
  "order": {
    "hops": [
      {
        "from": "Reader",
        "to": "Writer"
      }
    ]
  },
  "setting": {
    "errorLimit": {
      "record": "0"
    },
    "speed": {
      "throttle": true,
      "concurrent": 1,
      "mbps": "12"
    }
  },
  "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

Parâmetro	Descrição	Obrigatório	Valor padrão
`datasource`	Nome da fonte de dados Elasticsearch no DataWorks. Caso ainda não tenha sido criada, consulte Configurar uma fonte de dados Elasticsearch.	Sim	Nenhum
`index`	Nome do índice Elasticsearch de destino.	Sim	Nenhum
`indexType`	Tipo de índice do Elasticsearch.	Não	`Elasticsearch`
`cleanup`	Comportamento do índice antes da gravação. `true`: exclui o índice existente e o recria antes da gravação (todos os dados existentes são perdidos). `false`: mantém os dados existentes no índice.	Não	false
`batchSize`	Quantidade de documentos a inserir por lote.	Não	1.000
`trySize`	Número de tentativas após falha de gravação.	Não	30
`timeout`	Tempo limite do cliente em milissegundos.	Não	600.000
`discovery`	Modo de descoberta de nós. `true`: conecta-se a um nó aleatório do cluster e atualiza periodicamente a lista de servidores. `false`: conecta-se diretamente ao endpoint do cluster Elasticsearch.	Não	false
`compression`	Indica se os corpos das requisições HTTP devem ser compactados.	Não	true
`multiThread`	Indica se devem ser usadas múltiplas threads para requisições HTTP.	Não	true
`ignoreWriteError`	Se `true`, erros de gravação são ignorados sem nova tentativa e a tarefa continua.	Não	false
`ignoreParseError`	Se `true`, erros de análise de formato de dados são ignorados e a tarefa continua.	Não	true
`alias`	Alias a ser criado para o índice após a importação dos dados. Um alias funciona como uma view de banco de dados — operações no alias são aplicadas ao índice subjacente.	Não	Nenhum
`aliasMode`	Define como o alias é vinculado após a importação. `append`: adiciona o índice atual ao alias, que pode apontar para múltiplos índices. `exclusive`: remove primeiro o mapeamento de alias existente e depois adiciona o índice atual — o alias aponta para exatamente um índice. Aliases permitem migração de índices e consultas unificadas em múltiplos índices.	Não	`append`
`settings`	Configurações de índice aplicadas durante a criação do índice. Segue o formato padrão de configurações de índice do Elasticsearch.	Não	Nenhum
`column`	Definições de campos para os documentos a serem gravados. Cada campo requer, no mínimo, `name` e `type`. Consulte a referência de tipos de coluna abaixo.	Sim	Nenhum
`dynamic`	Controla se o mapeamento dinâmico do Elasticsearch é usado para campos não presentes em `column`. `true`: o Elasticsearch mapeia automaticamente campos não mapeados. `false`: os mapeamentos são gerados e atualizados apenas a partir da configuração de `column`. Para Elasticsearch 7.x, defina `"esVersion": "7"` no editor de código ao usar mapeamento dinâmico. Importante Ative `dynamic` apenas como solução alternativa para erros de mapeamento de campos. Essa opção pode gerar tipos de campo inesperados ou exceções de dados — avalie sua estrutura de dados antes de ativá-la.	Não	false
`actionType`	Tipo de ação de gravação. `index`: usa o `Index.Builder` do SDK do Elasticsearch. Se nenhum `_id` for especificado, o Elasticsearch gera um automaticamente. Se `_id` for especificado, o documento inteiro é substituído. `update`: atualiza campos específicos em um documento existente pelo `_id`. Se o `_id` não existir, o documento é inserido. Cada atualização recupera o documento completo para modificar campos específicos — isso pode afetar significativamente o desempenho. Se `actionType` for `update`, `primaryKeyInfo` é obrigatório.	Não	`index`
`primaryKeyInfo`	Define como o `_id` do documento é definido. Consulte a referência de configuração de chave primária abaixo.	Sim	`specific`
`esPartitionColumn`	Configuração de roteamento para gravações particionadas. Concatena os valores das colunas especificadas (com separador de string vazia) e define o resultado como valor de roteamento do Elasticsearch, direcionando documentos para um shard específico. Se não especificado, `_id` é usado como chave de roteamento para distribuir documentos uniformemente entre os shards.	Não	false
`enableWriteNull`	Controla se campos nulos da fonte são gravados no Elasticsearch. `true`: campos nulos são gravados; o campo correspondente no Elasticsearch será nulo. `false`: campos nulos não são gravados; o campo fica ausente no documento do Elasticsearch.	Não	true

Referência de tipos de coluna

O parâmetro column suporta os seguintes tipos de campo:

id          // Maps to _id in Elasticsearch. Documents with the same ID are overwritten, not re-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

Configuração específica por tipo:

text: suporta propriedades adicionais como analyzer, norms e index_options:

{
  "name": "col_text",
  "type": "text",
  "analyzer": "ik_max_word"
}

date: dois métodos para analisar dados da fonte:
- Método 1 (gravar como está): Defina "origin": true para gravar o valor do campo diretamente no Elasticsearch sem conversão. Configure também "format" para que o Writer defina o formato correto no mapeamento: ``json { "name": "col_date", "type": "date", "format": "yyyy-MM-dd HH:mm:ss", "origin": true } ``
- Método 2 (conversão de fuso horário): Omita origin e adicione "Timezone" para que o Data Integration converta o fuso horário antes da gravação: ``json { "name": "col_date", "type": "date", "format": "yyyy-MM-dd HH:mm:ss", "Timezone": "UTC" } ``

geo_shape: suporta tree (geohash ou quadtree) e precision:

{
  "name": "col_geo_shape",
  "type": "geo_shape",
  "tree": "quadtree",
  "precision": "10m"
}

other_params: use esta propriedade dentro de qualquer entrada column para definir propriedades de mapeamento do Elasticsearch não cobertas pelos tipos de campo padrão:
```
{
  "name": "guid",
  "type": "text",
  "other_params": {
    "doc_values": false
  }
}
```

Referência de configuração de chave primária

O parâmetro primaryKeyInfo suporta três modos:

Business primary key (`pk`): define _id com o valor de um campo específico:
```
"primaryKeyInfo": {
  "type": "pk",
  "column": ["id"]
}
```
Composite primary key (`specific`): define _id concatenando múltiplos valores de campo, separados por fieldDelimiter:

Na interface visual, a Primary key column configuration lista apenas campos que já existem no índice do Elasticsearch.
```
"primaryKeyInfo": {
  "type": "specific",
  "fieldDelimiter": ",",
  "column": ["col1", "col2"]
}
```
No primary key (`nopk`): o Elasticsearch gera _id automaticamente:
```
"primaryKeyInfo": {
  "type": "nopk"
}
```

Apêndice 2: Gravar dados no Elasticsearch em formato de array

Existem dois métodos disponíveis para gravar dados da fonte no Elasticsearch como um array.

Analisar como JSON

Se os dados da fonte já forem uma string de array JSON como "[1,2,3,4,5]", defina "json_array": true para analisá-la:

{
  "name": "docs_1",
  "type": "keyword",
  "json_array": true
}

Analisar com separador

Caso os dados da fonte sejam uma string delimitada como "1,2,3,4,5", defina "array": true na coluna e configure splitter no nível do parâmetro:

{
  "parameter": {
    "column": [
      {
        "name": "docs_2",
        "array": true,
        "type": "long"
      }
    ],
    "splitter": ","    // Must be at the same level as "column", not inside it.
  }
}

splitter é um parâmetro global — uma tarefa suporta apenas um separador. Se múltiplos campos de array usarem delimitadores diferentes (por exemplo, col1="1,2,3" e col2="6-7-8" ), não é possível configurar separadores distintos para cada campo.

Apêndice 3: Cenários

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

Extraia um documento inteiro do Elasticsearch como um único campo de string JSON. Defina "full": true no parâmetro do Reader.

Entrada — documento bruto do Elasticsearch:

"hits": [
  {
    "_index": "mutiltest_1",
    "_type": "_doc",
    "_id": "IXgdO4MB4GR_1DmrjTXP",
    "_score": 1.0,
    "_source": {
      "feature1": "value1",
      "feature2": "value2",
      "feature3": "value3"
    }
  }
]

Configuração do Reader:

"parameter": {
  "column": ["content"],
  "full": true
}

Saída — uma linha, uma coluna:

{"_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 objetos

Use notação de ponto e notação de índice de array para extrair propriedades específicas de campos object ou nested:

property — propriedade de nível superior
property.sub-property — subpropriedade aninhada
property[0].sub-property — elemento em um índice específico do array

A configuração para este cenário não está disponível na interface visual. Use o editor de código.

Configuração do Reader:

"multi": {
  "multi": true
}

Entrada — documento bruto do Elasticsearch:

"hits": [
  {
    "_index": "mutiltest_1",
    "_type": "_doc",
    "_id": "7XAOOoMB4GR_1Dmrrust",
    "_score": 1.0,
    "_source": {
      "level1": {
        "level2": [
          { "level3": "testlevel3_1" },
          { "level3": "testlevel3_2" }
        ]
      }
    }
  }
]

Configuração de coluna:

"column": [
  "level1",
  "level1.level2",
  "level1.level2[0]",
  "level1.level2.level3"
]

Saída — uma linha, quatro colunas:

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

Se um nó pai no caminho for um array, o resultado será null . Por exemplo, level1.level2.level3 retorna null porque level2 é um array. Use level1.level2[0].level3 ou level1.level2[1].level3 em vez disso. O curinga level1.level2[*].level3 não é suportado atualmente.

Chaves que contêm um ponto ( . ) não são suportadas. Por exemplo, {"level1.level2": {"level3": "value"}} retorna null para este caminho.

Cenário 3: Dividir propriedades de array em múltiplas linhas

Expanda um array um-para-muitos em linhas separadas usando o curinga [*] no caminho da coluna.

Configuração: property[*].sub-property

Na interface visual, defina Split multi-row array column name para gerar a configuração de script equivalente.

Configuração do Reader:

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

O valor de key deve ser um campo de array. Caso contrário, ocorrerá um erro.

Entrada — dois documentos do Elasticsearch:

[
  {
    "_source": {
      "headers": [
        { "remoteip": "192.0.2.1" },
        { "remoteip": "192.0.2.2" }
      ]
    }
  },
  {
    "_source": {
      "headers": [
        { "remoteip": "192.0.2.3" },
        { "remoteip": "192.0.2.4" }
      ]
    }
  }
]

Configuração de coluna:

"column": ["headers[*].remoteip"]

Saída — quatro linhas:

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

Deduplique e mescle um campo de array em uma única string separada por vírgulas. Inclua [] no nome da coluna para ativar a deduplicação. A deduplicação usa o resultado de toString como chave de comparação.

Configuração: property[]

Esta configuração não está disponível na interface visual. Use o editor de código.

Configuração do Reader:

"multi": {
  "multi": true
}

Entrada:

"_source": {
  "feature1": ["value1", "value1", "value2", "value2", "value3"]
}

Configuração de coluna:

"column": ["feature1[]"]

Saída — uma linha, uma coluna:

"value1,value2,value3"

Cenário 5: Mesclar e sincronizar múltiplas propriedades

Mescle múltiplas propriedades em uma única coluna. Todas as propriedades nomeadas são concatenadas em um único valor.

Configuração: property1,property2,...

Esta configuração não está disponível na interface visual. Use o editor de código.

Configuração do Reader:

"multi": {
  "multi": true
}

Entrada:

"_source": {
  "feature1": "feature1",
  "feature2": [1, 2, 3],
  "feature3": { "child": "feature3" }
}

Configuração de coluna:

"column": ["feature1,feature2,feature3"]

Saída — uma linha, uma coluna:

"feature1,[1,2,3],{"child":"feature3"}"

Cenário 6: Sincronizar seletivamente múltiplas propriedades

Retorne o primeiro valor não nulo de uma lista de propriedades. Se nenhuma das propriedades especificadas tiver valor, null será gravado.

Configuração: property1|property2|...

Esta configuração não está disponível na interface visual. Use o editor de código.

Configuração do Reader:

"multi": {
  "multi": true
}

Entrada:

"_source": {
  "feature1": "feature1",
  "feature2": [1, 2, 3],
  "feature3": { "child": "feature3" }
}

Configuração de coluna:

"column": ["feature1|feature2|feature3"]

Saída — uma linha, uma coluna (primeiro valor não nulo):

"feature1"

Referências

O Data Integration oferece suporte a fontes de dados adicionais. Para obter a lista completa, consulte Fontes de dados e sincronização.

Modo	Direção
Leitura offline (Elasticsearch Reader)	Elasticsearch → downstream
Gravação offline (Elasticsearch Writer)	Upstream → Elasticsearch
Gravação em tempo real	Upstream → Elasticsearch (streaming)