Antes de começar
-
Criar um índice do Elasticsearch
O DataHub permite sincronizar dados para um índice em uma instância do Elasticsearch. Esse recurso é compatível com as versões 5, 6 e 7 do Elasticsearch.
Atualmente, o DataHub sincroniza apenas dados de tópicos do tipo TUPLE para o Elasticsearch. Antes de iniciar a tarefa de sincronização, certifique-se de ter criado um índice no Elasticsearch ou de que a criação automática de índices esteja ativada. Caso contrário, a tarefa de sincronização falhará.
-
Preparar e autorizar uma conta para a tarefa de sincronização
Para criar uma tarefa de sincronização do Elasticsearch, forneça o endpoint do Elasticsearch, os detalhes do índice e as credenciais da conta. Verifique se as informações da conta são válidas para evitar falhas na criação da tarefa.
Criar uma tarefa de sincronização
Acesse a página .
No canto superior direito, clique em
+ Sync.Selecione
Elasticsearchcomo o tipo de conector.
Parâmetros
Endpoint
O endereço do service Elasticsearch requer um endereço interno e uma porta interna no formato endereço interno:porta interna.
Por exemplo, se o endereço interno for es-cn-xxx.elasticsearch.aliyuncs.com e a porta interna for 9200, insira es-cn-xxx.elasticsearch.aliyuncs.com:9200.
Index
É possível especificar o índice de destino de duas formas: estática ou dinâmica.
Índice estático
Todos os dados são gravados em um único índice predefinido. Crie esse índice antecipadamente ou ative a criação automática de índices no Elasticsearch.
Índice dinâmico
Os dados são gravados em índices diferentes com base em um período de tempo ou nos valores de uma coluna de dados específica. Para que a indexação dinâmica funcione, ative a criação automática de índices no Elasticsearch. Selecione até uma coluna para gerar o nome do índice.
Formatos de hora suportados:
|
Ano |
Mês |
Dia |
Semana |
|
%Y |
%m |
%d |
%U |
Exemplo 1: Gerar um novo índice todos os dias à meia-noite Configure o índice como
test_${%Y-%m-%d}. Se a data atual for 31 de março de 2021, o índice final serátest_2021-03-31.Exemplo 2: Gerar um novo índice a partir de uma coluna de dados Os dados contêm uma coluna chamada
col1, e o índice está configurado comotest_${col1}. Se dois registros de dados tiveremAAAeBBBcomo valores para a coluna col1, os registros serão gravados nos índicestest_AAAetest_BBB, respectivamente.
À medida que o número de índices no Elasticsearch aumenta, o desempenho de gravação pode diminuir. Um número excessivo de índices pode causar tempos limite de gravação no DataHub. Ao usar um índice dinâmico, planeje seu padrão de nomenclatura para evitar a criação de uma quantidade excessiva de índices.
User/Password
Nome de usuário e senha para acessar o Elasticsearch.
Type column
O DataHub gera o tipo de documento de maneira diferente dependendo da versão do Elasticsearch. No Elasticsearch 5, é possível criar vários tipos dentro de um único índice. No Elasticsearch 6, apenas um tipo por índice é permitido. Esta configuração não pode estar vazia.
No Elasticsearch 5, o DataHub usa o valor da coluna Type selecionada como o tipo de documento. Se você selecionar várias colunas, seus valores serão concatenados com um separador "|" para formar o tipo. Os campos selecionados para a coluna Type não podem conter valores nulos.
No Elasticsearch 6, o DataHub usa o nome da coluna selecionada como o tipo de documento. Se você selecionar várias colunas, seus nomes serão concatenados com um separador "|". O Elasticsearch 6 aceita qualquer string como nome de tipo.
Não é possível especificar um nome de tipo personalizado no console ao criar uma tarefa de sincronização para o Elasticsearch 6. Para usar um nome de tipo personalizado, crie a tarefa usando o SDK.
No Elasticsearch 7, todos os documentos usam um tipo padrão, portanto, não é necessário selecionar uma coluna Type.
Exemplo:
DataHub Schema : f1 string, f2 string, f3 string, f4 string
Data record : ["test1","test2","test3",null]|
Coluna Type |
Tipo no Elasticsearch 5 |
Tipo no Elasticsearch 6 |
||
|
f1 |
test1 |
f1 |
||
|
f1,f3 |
test1 |
test3 |
f1 |
f3 |
|
ff |
Falha na criação |
ff |
||
|
f1,ff |
Falha na criação |
f1 |
ff |
|
|
f4 |
Criação bem-sucedida, mas falha na sincronização (dados incorretos) |
Criação e sincronização bem-sucedidas |
ID column
É possível gerar um ID de documento para o Elasticsearch a partir dos dados de registro do DataHub. Se nenhuma coluna for selecionada, o Elasticsearch gerará um ID exclusivo para cada documento. Ao selecionar uma ou mais colunas, seus valores são concatenados com um separador "|" para criar o ID do documento. Os campos selecionados para a coluna ID não podem conter valores nulos.
Exemplo:
DataHub Schema : f1 string, f2 string, f3 string, f4 string
Data record : ["test1","test2","test3",null]|
Coluna Id |
Id do documento |
|
|
O Elasticsearch gera automaticamente um ID exclusivo |
||
|
f1 |
test1 |
|
|
f1,f3 |
test1 |
test3 |
|
ff |
Falha na criação |
|
|
f4 |
Criação bem-sucedida, mas falha na sincronização (dados incorretos) |
Router column
Você pode gerar um valor de roteamento para o Elasticsearch a partir dos dados de registro do DataHub. Se nenhuma coluna for selecionada, o recurso de roteamento do Elasticsearch não será utilizado. Ao selecionar uma ou mais colunas, seus valores serão usados como valor de roteamento. Se várias colunas forem selecionadas, seus valores serão concatenados com um separador "|". Os campos selecionados para a coluna Router não podem conter valores nulos.
Para ver um exemplo, consulte a ID column.
Imported fields
Selecione os campos do tópico do DataHub para sincronizar com o Elasticsearch. O DataHub não sincroniza campos não selecionados. No Elasticsearch 5, os dados finais do documento não incluem os campos usados para a coluna ID e a coluna Type. Para obter mais informações sobre como os dados são transformados, consulte os exemplos nas seções a seguir.
Network type
Selecione o tipo de rede com base na implantação da sua instância do Elasticsearch. As instâncias do Alibaba Cloud Elasticsearch na cloud pública executam dentro de uma VPC. Portanto, selecione VPC como o tipo de rede para essas tarefas de sincronização. Ao usar o tipo de rede VPC, forneça o ID da VPC e o ID da instância.
Ao inserir o ID da instância, adicione o sufixo -worker. Por exemplo, se o ID da sua instância for es-cn-xxx, insira es-cn-xxx-worker.
Exemplos de gravação de dados
Os exemplos a seguir pressupõem que a tarefa de sincronização do Elasticsearch foi criada com êxito. Se a criação da tarefa falhar, revise e corrija sua configuração.
Schema do DataHub:
|
Nome do campo |
Tipo do campo |
|
f1 |
BIGINT |
|
f2 |
STRING |
|
f3 |
BOOLEAN |
|
f4 |
DOUBLE |
|
f5 |
TIMESTAMP |
|
f6 |
DECIMAL |
-
Exemplo 1
Coluna Type = f1 (Não aplicável ao Elasticsearch 7)
Coluna ID = f2
-
Campos importados = f1,f2,f3,f4,f5,f6
-
Registro de dados = v1,v2,v3,v4,v5,v6
Versão do ES
Tipo
Id
Dados
ES5
v1
v2
{f3:v3,f4:v4,f5:v5,f6:v6}ES6
f1
v2
{f1:v1,f3:v3,f4:v4,f5:v5,f6:v6}ES7
-
v2
{f1:v1,f3:v3,f4:v4,f5:v5,f6:v6} -
Registro de dados = null,v2,v3,v4,v5,v6
Versão do ES
Tipo
Id
Dados
ES5
-
-
Um valor nulo na coluna Type resulta em dados incorretos.
ES6
f1
v2
{f1:v1,f3:v3,f4:v4,f5:v5,f6:v6}ES7
-
v2
{f1:v1,f3:v3,f4:v4,f5:v5,f6:v6} -
Registro de dados = v1,null,v3,v4,v5,v6
Versão do ES
Tipo
Id
Dados
ES5
-
-
Um valor nulo na coluna ID resulta em dados incorretos.
ES6
-
-
Um valor nulo na coluna ID resulta em dados incorretos.
ES7
-
-
Um valor nulo na coluna ID resulta em dados incorretos.
-
-
Exemplo 2
Coluna Type = f1,f2 (Não aplicável ao Elasticsearch 7)
Coluna ID = f3,f4
-
Campos importados = f1,f2,f3,f4,f5,f6
-
Registro de dados = v1,v2,v3,v4,v5,v6
Versão do ES
Tipo
Id
Dados
ES5
v1
v2
v3
v4
{f5:v5,f6:v6}ES6
f1
f2
v3
v4
{f1:v1,f2:v2,f5:v5,f6:v6}
ES7
-
v3
v4
{f1:v1,f2:v2,f5:v5,f6:v6}
-
Registro de dados = v1,null,v3,v4,v5,v6
Versão do ES
Tipo
Id
Dados
ES5
-
-
Um valor nulo na coluna Type resulta em dados incorretos.
ES6
f1
f2
v3
v4
{f1:v1,f2:v2,f5:v5,f6:v6}
ES7
-
v3
v4
{f1:v1,f2:v2,f5:v5,f6:v6}
-
Registro de dados = v1,v2,null,v4,v5,v6
Versão do ES
Tipo
Id
Dados
ES5
-
-
Um valor nulo na coluna ID resulta em dados incorretos.
ES6
-
-
Um valor nulo na coluna ID resulta em dados incorretos.
ES7
-
-
Um valor nulo na coluna ID resulta em dados incorretos.
-
-
Exemplo 3
Coluna Type = f1 (Não aplicável ao Elasticsearch 7)
Coluna ID = f2
Coluna Router = f3
-
Campos importados = f1,f2,f3,f4,f5,f6
-
Registro de dados = v1,v2,v3,v4,v5,v6
Versão do ES
Tipo
Id
Router
Dados
ES5
v1
v2
v3
{f4:v4,f5:v5,f6:v6}ES6
f1
v2
v3
{f1:v1,f4:v4,f5:v5,f6:v6}ES7
-
v2
v3
{f1:v1,f4:v4,f5:v5,f6:v6} -
Registro de dados = null,v2,v3,v4,v5,v6
Versão do ES
Tipo
Id
Dados
ES5
-
-
Um valor nulo na coluna Type resulta em dados incorretos.
ES6
f1
v2
{f1:v1,f4:v4,f5:v5,f6:v6}ES7
-
v2
{f1:v1,f4:v4,f5:v5,f6:v6} -
Registro de dados = v1,null,v3,v4,v5,v6
Versão do ES
Tipo
Id
Dados
ES5
-
-
Um valor nulo na coluna ID resulta em dados incorretos.
ES6
-
-
Um valor nulo na coluna ID resulta em dados incorretos.
ES7
-
-
Um valor nulo na coluna ID resulta em dados incorretos.
-
Registro de dados = v1,v2,null,v4,v5,v6
Versão do ES
Tipo
Id
Dados
ES5
-
-
Um valor nulo na coluna Router resulta em dados incorretos.
ES6
-
-
Um valor nulo na coluna Router resulta em dados incorretos.
ES7
-
-
Um valor nulo na coluna Router resulta em dados incorretos.
-
-
Criar um índice do Elasticsearch
Por padrão, o Elasticsearch cria automaticamente um índice, então geralmente você pode pular esta etapa. Se a criação automática de índices estiver desativada, crie o índice manualmente. Para comandos específicos, consulte official Elasticsearch documentation.
-
Criar um tópico do DataHub
Apenas tópicos do tipo TUPLE suportam sincronização com o Elasticsearch. Para obter instruções sobre como criar um tópico do DataHub, consulte Topic Operations.
-
Criar a tarefa de sincronização do Elasticsearch
Neste exemplo, ao criar a tarefa de sincronização, usamos f1 e f2 como coluna Type, f3 e f4 como coluna ID e selecionamos todos os campos para importação.
-
Gravar dados no tópico do DataHub
Use o SDK do DataHub ou um plugin para gravar dados. Após gravar um registro, faça uma amostragem dos dados no console para verificar o que foi gravado.
-
Verificar a sincronização de dados
Primeiro, verifique o checkpoint da tarefa de sincronização do Elasticsearch. Uma alteração no checkpoint e no tempo de sincronização indica que os dados foram sincronizados. O tempo de sincronização reflete quando os dados foram gravados no DataHub, e um valor de checkpoint igual a 1 significa que o primeiro registro de dados (no índice 0) foi gravado.
Em seguida, verifique os dados no Elasticsearch. Use o Kibana para confirmar que os dados foram sincronizados com êxito.
Gerenciar a tarefa de sincronização
Na página de detalhes do conector, visualize o status de execução da tarefa, o checkpoint e outras informações. Também é possível reiniciar e parar a tarefa. Pare a tarefa antes de redefinir o checkpoint.
Exemplo de ponta a ponta
Este exemplo usa o Alibaba Cloud Elasticsearch 6.7 para demonstrar o processo completo de sincronização de dados do DataHub para o Elasticsearch. As operações relacionadas ao Elasticsearch são executadas usando o Kibana Dev Tools. Para outros métodos, consulte official Elasticsearch documentation.