Elasticsearch:Usar o ES-Hadoop para gravar dados do HDFS no Elasticsearch

Procedimento

1. Preparations

   Crie uma instância do Alibaba Cloud Elasticsearch e uma instância do E-MapReduce (EMR) na mesma virtual private cloud (VPC). Em seguida, ative o recurso de criação automática de índice para a instância do Elasticsearch e prepare os dados de teste e um ambiente Java.

2. Step 1: Upload the ES-Hadoop JAR package to HDFS

   Baixe o pacote de instalação do ES-Hadoop e envie-o para o diretório HDFS no nó mestre do cluster EMR.

3. Step 2: Configure pom dependencies

   Crie um projeto Java Maven e configure as dependências pom.

4. Step 3: Write and run the MapReduce task

   Escreva o código Java da tarefa MapReduce para gravar dados no Elasticsearch. Empacote o código em um arquivo JAR, envie-o para o cluster EMR e execute-o para concluir a gravação dos dados.

5. Step 4: Verify the results

   No console Kibana da instância do Elasticsearch, visualize os dados gravados pela tarefa MapReduce.

Preparações

1. Crie uma instância do Alibaba Cloud Elasticsearch e ative o recurso de criação automática de índice.

   Para obter mais informações, consulte Create an Alibaba Cloud Elasticsearch instance e Configure YML parameters. Este tópico usa uma instância do Elasticsearch 6.7.0 como exemplo.

   Importante

   Em ambientes de produção, desative o recurso de criação automática de índice. Crie o índice e o mapeamento antecipadamente. Como este tópico serve apenas para fins de teste, o recurso de criação automática de índice está ativado.

2. Crie uma instância EMR na mesma VPC da instância do Elasticsearch.

   Use as seguintes configurações de instância:

   • Edição: EMR-3.29.0

   • Serviços obrigatórios: HDFS 2.8.5. Mantenha as configurações padrão para os demais serviços.

   Para obter mais informações, consulte Create a cluster.

   Importante

   A lista de permissões de endereços IP privados de um cluster Elasticsearch é definida como 0.0.0.0/0 por padrão. Visualize essa configuração na página de configuração de segurança. Caso altere esse padrão, adicione o endereço IP interno do cluster EMR à lista de permissões:

   • Consulte View cluster list and details para obter o endereço IP interno do cluster EMR.

   • Consulte Configure a public or private IP address whitelist for an Elasticsearch cluster para configurar a lista de permissões de endereços IP privados da VPC para o cluster Elasticsearch.

3. Prepare os dados de teste em JSON, grave-os no arquivo map.json e envie o arquivo para o diretório /tmp/hadoop-es no HDFS.

   Este tópico utiliza os seguintes dados de teste.

   {"id": 1, "name": "zhangsan", "birth": "1990-01-01", "addr": "No.969, wenyixi Rd, yuhang, hangzhou"}
   {"id": 2, "name": "lisi", "birth": "1991-01-01", "addr": "No.556, xixi Rd, xihu, hangzhou"}
   {"id": 3, "name": "wangwu", "birth": "1992-01-01", "addr": "No.699 wangshang Rd, binjiang, hangzhou"}

4. Prepare um ambiente Java. A versão do JDK deve ser 1.8.0 ou posterior.

Etapa 1: Enviar o pacote JAR do ES-Hadoop para o HDFS

1. Baixe o pacote de instalação do ES-Hadoop cuja versão corresponda à do cluster Elasticsearch.

   Este tópico utiliza o arquivo elasticsearch-hadoop-6.7.0.zip.

2. Faça login no console EMR, obtenha o endereço IP do nó mestre e use SSH para fazer login na instância ECS correspondente.

   Para obter mais informações, consulte Log on to a cluster.

3. Envie o pacote elasticsearch-hadoop-6.7.0.zip para o nó mestre e descompacte-o para obter o arquivo elasticsearch-hadoop-6.7.0.jar.

4. Crie um diretório HDFS e envie o arquivo elasticsearch-hadoop-6.7.0.jar para esse diretório.

   hadoop fs -mkdir /tmp/hadoop-es
   hadoop fs -put elasticsearch-hadoop-6.7.0/dist/elasticsearch-hadoop-6.7.0.jar /tmp/hadoop-es

Etapa 2: Configurar dependências pom

Crie um projeto Java Maven e adicione as seguintes dependências pom ao arquivo pom.xml do projeto.

<build>
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-shade-plugin</artifactId>
            <version>2.4.1</version>
            <executions>
                <execution>
                    <phase>package</phase>
                    <goals>
                        <goal>shade</goal>
                    </goals>
                    <configuration>
                        <transformers>
                            <transformer
                                    implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                <mainClass>WriteToEsWithMR</mainClass>
                            </transformer>
                        </transformers>
                    </configuration>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

<dependencies>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-hdfs</artifactId>
        <version>2.8.5</version>
    </dependency>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-mapreduce-client-jobclient</artifactId>
        <version>2.8.5</version>
    </dependency>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-common</artifactId>
        <version>2.8.5</version>
    </dependency>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-auth</artifactId>
        <version>2.8.5</version>
    </dependency>

    <dependency>
        <groupId>org.elasticsearch</groupId>
        <artifactId>elasticsearch-hadoop-mr</artifactId>
        <version>6.7.0</version>
    </dependency>

    <dependency>
        <groupId>commons-httpclient</groupId>
        <artifactId>commons-httpclient</artifactId>
        <version>3.1</version>
    </dependency>
</dependencies>

Etapa 3: Escrever e executar a tarefa MapReduce

1. Escreva o código de exemplo.

   O código a seguir lê os arquivos JSON do diretório /tmp/hadoop-es no HDFS. Em seguida, grava cada linha desses arquivos como um documento no Elasticsearch. A classe EsOutputFormat conclui a operação de gravação na fase Map.

   import java.io.IOException;
   import org.apache.hadoop.conf.Configuration;
   import org.apache.hadoop.conf.Configured;
   import org.apache.hadoop.fs.Path;
   import org.apache.hadoop.io.NullWritable;
   import org.apache.hadoop.io.Text;
   import org.apache.hadoop.mapreduce.Job;
   import org.apache.hadoop.mapreduce.Mapper;
   import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
   import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
   import org.apache.hadoop.util.GenericOptionsParser;
   import org.elasticsearch.hadoop.mr.EsOutputFormat;
   import org.apache.hadoop.util.Tool;
   import org.apache.hadoop.util.ToolRunner;
   
   public class WriteToEsWithMR extends Configured implements Tool {
   
       public static class EsMapper extends Mapper<Object, Text, NullWritable, Text> {
           private Text doc = new Text();
   
           @Override
           protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
               if (value.getLength() > 0) {
                   doc.set(value);
                   System.out.println(value);
                   context.write(NullWritable.get(), doc);
               }
           }
       }
       public int run(String[] args) throws Exception {
           Configuration conf = new Configuration();
           String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
   
           conf.setBoolean("mapreduce.map.speculative", false);
           conf.setBoolean("mapreduce.reduce.speculative", false);
           conf.set("es.nodes", "es-cn-4591jumei000u****.elasticsearch.aliyuncs.com");
           conf.set("es.port","9200");
           conf.set("es.net.http.auth.user", "elastic");
           conf.set("es.net.http.auth.pass", "xxxxxx");
           conf.set("es.nodes.wan.only", "true");
           conf.set("es.nodes.discovery","false");
           conf.set("es.input.use.sliced.partitions","false");
           conf.set("es.resource", "maptest/_doc");
           conf.set("es.input.json", "true");
   
           Job job = Job.getInstance(conf);
           job.setInputFormatClass(TextInputFormat.class);
           job.setOutputFormatClass(EsOutputFormat.class);
           job.setMapOutputKeyClass(NullWritable.class);
           job.setMapOutputValueClass(Text.class);
           job.setJarByClass(WriteToEsWithMR.class);
           job.setMapperClass(EsMapper.class);
   
           FileInputFormat.setInputPaths(job, new Path(otherArgs[0]));
   
           return job.waitForCompletion(true) ? 0 : 1;
       }
   
       public static void main(String[] args) throws Exception {
           int ret = ToolRunner.run(new WriteToEsWithMR(), args);
           System.exit(ret);
       }
   }

   Tabela 1. Descrições dos parâmetros do ES-Hadoop

   Parâmetro	Valor padrão	Descrição
   es.nodes	localhost	Endpoint da instância do Alibaba Cloud Elasticsearch. Use o endpoint privado, disponível na página Basic Information da instância. Para obter mais informações, consulte View the basic information of an instance.
   es.port	9200	Número da porta da instância do Elasticsearch.
   es.net.http.auth.user	elastic	Nome de usuário para acessar a instância do Elasticsearch. Nota Se você especificar a conta elastic na aplicação, qualquer alteração subsequente de senha dessa conta poderá causar interrupções temporárias no serviço devido ao atraso de propagação. Portanto, não recomendamos o uso da conta elastic. Em vez disso, crie um usuário dedicado com as permissões apropriadas no console Kibana. Para obter mais informações, consulte Use the RBAC mechanism of Elasticsearch X-Pack to control user access.
   es.net.http.auth.pass	/	Senha para acessar a instância do Elasticsearch.
   es.nodes.wan.only	false	Define se o sniffing de nós deve ser executado ao conectar-se a um cluster Elasticsearch que utiliza um endereço IP virtual na nuvem. true: Ativado false: Parâmetro não definido.
   es.nodes.discovery	true	Define se a descoberta de nós deve ser desativada. true: Desativar false: Não desativado
   es.input.use.sliced.partitions	true	Define se partições fatiadas devem ser usadas. true: Usa partições fatiadas. Definir este parâmetro como true pode aumentar significativamente o tempo de pré-leitura do índice, tornando-o às vezes muito maior que o próprio tempo de consulta. Recomendamos definir este parâmetro como false para melhorar o desempenho da consulta. false: Não usa partições fatiadas.
   es.index.auto.create	true	Define se um índice deve ser criado automaticamente caso não exista durante a gravação de dados de um componente Hadoop em um cluster Elasticsearch. true: Criar o índice automaticamente. false: Criação automática desativada.
   es.resource	/	Índice e tipo de origem ou destino da leitura/gravação.
   es.input.json	false	Define se a entrada está no formato JSON. true: Entrada no formato JSON. false: Entrada não está no formato JSON.
   es.mapping.names	/	Mapeamento entre os campos da tabela e os campos do índice do Elasticsearch.
   es.read.metadata	false	Ative esta propriedade se a operação envolver campos internos do Elasticsearch, como _id.

   Para obter mais informações sobre os itens de configuração do ES-Hadoop, consulte a documentação oficial de configuração.

2. Empacote o código em um arquivo JAR e envie-o para uma máquina cliente do EMR, como um gateway ou o nó mestre do cluster EMR.

3. Na máquina cliente do EMR, execute o seguinte comando para rodar o programa MapReduce.

   hadoop jar es-mapreduce-1.0-SNAPSHOT.jar /tmp/hadoop-es/map.json

   Nota

   Substitua es-mapreduce-1.0-SNAPSHOT.jar pelo nome do arquivo JAR enviado.

Etapa 4: Verificar os resultados

1. Faça login no console Kibana da instância do Alibaba Cloud Elasticsearch.

   Para obter mais informações, consulte Log in to the Kibana console.

2. No painel de navegação à esquerda, clique em Dev Tools.

3. Na aba Console, execute o seguinte comando para visualizar os dados gravados pela tarefa MapReduce.

   GET maptest/_search
   {
     "query": {
       "match_all": {}
     }
   }

   Uma consulta bem-sucedida retorna o seguinte resultado.

   {
     "took" : 8,
     "timed_out" : false,
     "_shards" : {
       "total" : 5,
       "successful" : 5,
       "skipped" : 0,
       "failed" : 0
     },
     "hits" : {
       "total" : 3,
       "max_score" : 1.0,
       "hits" : [
         {
           "_index" : "maptest",
           "_type" : "_doc",
           "_id" : "V8D0KnUB0sZ7Mms_YRGu",
           "_score" : 1.0,
           "_source" : {
             "id" : 3,
             "name" : "wangwu",
             "birth" : "1992-01-01",
             "addr" : "No.699 wangshang Rd, binjiang, hangzhou"
           }
         },
         {
           "_index" : "maptest",
           "_type" : "_doc",
           "_id" : "WMD0KnUB0sZ7Mms_YRGu",
           "_score" : 1.0,
           "_source" : {
             "id" : 2,
             "name" : "lisi",
             "birth" : "1991-01-01",
             "addr" : "No.556, xixi Rd, xihu, hangzhou"
           }
         },
         {
           "_index" : "maptest",
           "_type" : "_doc",
           "_id" : "WcD0KnUB0sZ7Mms_YRGu",
           "_score" : 1.0,
           "_source" : {
             "id" : 1,
             "name" : "zhangsan",
             "birth" : "1990-01-01",
             "addr" : "No.969, wenyixi Rd, yuhang, hangzhou"
           }
         }
       ]
     }
   }

Resumo

Este tópico demonstra como usar o ES-Hadoop para gravar dados no Elasticsearch com uma tarefa MapReduce, utilizando o Alibaba Cloud Elasticsearch e o EMR como exemplo. Também é possível usar uma tarefa MapReduce para consultar dados no Elasticsearch. A configuração para consulta é semelhante à de gravação. Para obter mais informações, consulte a documentação oficial sobre Leitura de dados do Elasticsearch.