Elasticsearch:Use ES-Hadoop to ative Apache Spark to write data to and read data from Alibaba Cloud Elasticsearch

Pré-requisitos

Antes de começar, verifique se você tem:

• Um cluster do Alibaba Cloud Elasticsearch (este tutorial usa a V6.7.0) com o recurso Auto Indexing ativado

  Importante

  Desative o recurso Auto Indexing em ambientes de produção. Crie o índice e configure os mappings antecipadamente. Este tutorial ativa o Auto Indexing apenas para fins de teste.

• Um cluster E-MapReduce (EMR) executando na mesma virtual private cloud (VPC) do cluster Elasticsearch, com a seguinte configuração:

  • Versão do EMR: EMR-3.29.0

  • Serviços necessários: Spark 2.4.5 (mantenha as configurações padrão para os demais serviços)

• O endereço IP privado do cluster EMR adicionado à lista de permissões de endereços IP privados do cluster Elasticsearch, caso você tenha alterado a lista de permissões padrão (o padrão é 0.0.0.0/0). Para obter o endereço IP privado do cluster EMR, consulte Visualize the cluster list and cluster details. Para atualizar a lista de permissões, consulte Configure a public or private IP address whitelist for an Elasticsearch cluster.

• JDK 1.8.0 ou posterior

Para obter instruções sobre como criar um cluster Elasticsearch, consulte Crie an Alibaba Cloud Elasticsearch cluster e Access and configure an Elasticsearch cluster. Para obter instruções sobre como criar um cluster EMR, consulte Crie a cluster.

Como funciona

1. Carregue os dados de teste JSON no Hadoop Distributed File System (HDFS) no nó mestre do EMR.

2. Crie um projeto Java Maven com as dependências do ES-Hadoop e do Spark e compile as classes de gravação e leitura.

3. Empacote o código compilado em um JAR e envie-o como um job do Spark usando spark-submit.

4. O ES-Hadoop serializa cada registro do Resilient Distributed Dataset (RDD) e o grava no índice do Elasticsearch especificado por meio da API REST.

5. Verifique os dados gravados executando uma consulta no console Dev Tools do Kibana.

Preparar dados de teste

1. Faça login no console do E-MapReduce, obtenha o endereço IP do nó mestre do EMR e conecte-se via SSH à instância ECS correspondente. Para mais detalhes, consulte Log on to a cluster.

2. Crie um arquivo chamado http_log.txt com os seguintes registros JSON:

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

3. Carregue o arquivo no HDFS:

   hadoop fs -put http_log.txt /tmp/hadoop-es

Adicionar dependências POM

Crie um projeto Java Maven e adicione as seguintes dependências ao arquivo pom.xml. Certifique-se de que cada versão corresponda à versão do service Alibaba Cloud associado. Por exemplo, elasticsearch-spark-20_2.11 deve corresponder à versão do seu cluster Elasticsearch, e spark-core_2.12 deve corresponder à versão do seu HDFS.

<dependencies>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.12</artifactId>
        <version>2.4.5</version>
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-sql_2.11</artifactId>
        <version>2.4.5</version>
    </dependency>
    <dependency>
        <groupId>org.elasticsearch</groupId>
        <artifactId>elasticsearch-spark-20_2.11</artifactId>
        <version>6.7.0</version>
    </dependency>
</dependencies>

Gravar dados no Elasticsearch

O código Java a seguir lê o arquivo http_log.txt do HDFS e grava cada linha como um documento no índice company/_doc. Substitua os valores de espaço reservado pelo endpoint e pelas credenciais reais do seu cluster.

import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.elasticsearch.spark.rdd.api.java.JavaEsSpark;
import org.spark_project.guava.collect.ImmutableMap;

public class SparkWriteEs {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf();
        conf.setAppName("Es-write");
        conf.set("es.nodes", "es-cn-n6w1o1x0w001c****.elasticsearch.aliyuncs.com"); // (1)
        conf.set("es.net.http.auth.user", "elastic");
        conf.set("es.net.http.auth.pass", "xxxxxx");                                 // (2)
        conf.set("es.nodes.wan.only", "true");                                       // (3)
        conf.set("es.nodes.discovery", "false");                                     // (4)
        conf.set("es.input.use.sliced.partitions", "false");                         // (5)

        SparkSession ss = new SparkSession(new SparkContext(conf));
        final AtomicInteger employeesNo = new AtomicInteger(0);

        JavaRDD<Map<Object, ?>> javaRDD = ss.read().text("/tmp/hadoop-es/http_log.txt") // (6)
                .javaRDD().map((Function<Row, Map<Object, ?>>) row ->
                        ImmutableMap.of("employees", employeesNo.getAndAdd(1), row.mkString()));

        JavaEsSpark.saveToEs(javaRDD, "company/_doc");
    }
}

1. es.nodes — o endpoint interno do seu cluster Elasticsearch. Obtenha-o na página Basic Information do cluster. Consulte Visualize the basic information of a cluster.

2. es.net.http.auth.pass — a senha definida durante a criação do cluster. Substitua xxxxxx pela sua senha real.

3. es.nodes.wan.only — defina como true porque o Alibaba Cloud Elasticsearch utiliza um IP virtual. Isso desativa a descoberta de nós e direciona todo o tráfego pelo endpoint definido em es.nodes.

4. es.nodes.discovery — deve ser false para o Alibaba Cloud Elasticsearch. Os endereços internos dos nós do cluster não são acessíveis externamente, portanto o mecanismo nativo de descoberta de nós não funciona.

5. es.input.use.sliced.partitions — defina como false para ignorar a fase de leitura antecipada do índice, o que melhora a eficiência da consulta. A fase de leitura antecipada pode levar mais tempo do que a própria consulta de dados.

6. Substitua /tmp/hadoop-es/http_log.txt pelo caminho real dos seus dados de teste no HDFS.

Ler dados do Elasticsearch

O código a seguir lê todos os documentos do índice company/_doc e os imprime na saída padrão (stdout):

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.elasticsearch.spark.rdd.api.java.JavaEsSpark;
import java.util.Map;

public class ReadES {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf()
                .setAppName("readEs")
                .setMaster("local[*]")
                .set("es.nodes", "es-cn-n6w1o1x0w001c****.elasticsearch.aliyuncs.com")
                .set("es.port", "9200")
                .set("es.net.http.auth.user", "elastic")
                .set("es.net.http.auth.pass", "xxxxxx")
                .set("es.nodes.wan.only", "true")
                .set("es.nodes.discovery", "false")
                .set("es.input.use.sliced.partitions", "false")
                .set("es.resource", "company/_doc")
                .set("es.scroll.size", "500");

        JavaSparkContext sc = new JavaSparkContext(conf);
        JavaPairRDD<String, Map<String, Object>> rdd = JavaEsSpark.esRDD(sc);

        for (Map<String, Object> item : rdd.values().collect()) {
            System.out.println(item);
        }

        sc.stop();
    }
}

Enviar os jobs do Spark

1. Empacote o código compilado em um JAR e carregue-o no nó mestre do EMR ou em um cluster de gateway associado.

2. No cliente do EMR, execute os comandos a seguir para enviar os jobs.

   Gravar dados:

   cd /usr/lib/spark-current
   ./bin/spark-submit --master yarn --executor-cores 1 --class "SparkWriteEs" /usr/local/spark_es.jar

   Ler dados:

   cd /usr/lib/spark-current
   ./bin/spark-submit --master yarn --executor-cores 1 --class "ReadES" /usr/local/spark_es.jar

   Importante

   Substitua /usr/local/spark_es.jar pelo caminho real onde você carregou o JAR.

   Após a conclusão do job de leitura, a saída exibe cada documento recuperado do índice do Elasticsearch:

   20/11/09 14:52:19 INFO [Executor task launch worker for task 0] Executor: Adding file:/tmp/spark-383278cd-c582-4c1c-xxx                    /userFiles-8d64xxx
   20/11/09 14:52:19 INFO [Executor task launch worker for task 0] Executor: Finished task 0.0 in stage 0.0 (TID 0). 1274 bytes result sent to driver
   20/11/09 14:52:19 INFO [task-result-getter-0] TaskSetManager: Finished task 0.0 in stage 0.0 (TID 0) in 544 ms on localhost (executor driver) (1/1)
   20/11/09 14:52:19 INFO [task-result-getter-0] TaskSchedulerImpl: Removed TaskSet 0.0, whose tasks have all completed, from pool
   20/11/09 14:52:19 INFO [dag-scheduler-event-loop] DAGScheduler: ResultStage 0 (collect at ReadES.java:26) finished in 0.691 s
   20/11/09 14:52:19 INFO [main] DAGScheduler: Job 0 finished: collect at ReadES.java:26, took 0.761197 s
   {employees0={"id": 1, "name": "zhangsan", "birth": "1990-01-01", "addr": "No.969, wenyixi Rd, yuhang, hangzhou"}}
   {employees1={"id": 2, "name": "lisi", "birth": "1991-01-01", "addr": "No.556, xixi Rd, xihu, hangzhou"}}
   {employees2={"id": 3, "name": "wangwu", "birth": "1992-01-01", "addr": "No.699 wangshang Rd, binjiang, hangzhou"}}
   20/11/09 14:52:19 INFO [main] SparkUI: Stopped Spark Web UI at http://xxx:4041
   20/11/09 14:52:19 INFO [dispatcher-event-loop-1] MapOutputTrackerMasterEndpoint: MapOutputTrackerMasterEndpoint stopped!
   20/11/09 14:52:19 INFO [main] MemoryStore: MemoryStore cleared
   20/11/09 14:52:19 INFO [main] BlockManager: BlockManager stopped
   20/11/09 14:52:19 INFO [main] BlockManagerMaster: BlockManagerMaster stopped
   20/11/09 14:52:19 INFO [dispatcher-event-loop-1] OutputCommitCoordinator$OutputCommitCoordinatorEndpoint: OutputCommitCoordinator stopped!
   20/11/09 14:52:19 INFO [main] SparkContext: Successfully stopped SparkContext
   20/11/09 14:52:19 INFO [pool-1-thread-1] ShutdownHookManager: Shutdown hook called
   20/11/09 14:52:19 INFO [pool-1-thread-1] ShutdownHookManager: Deleting directory /tmp/spark-383278cd-c582-4c1c-b4b6-xxx
   20/11/09 14:52:19 INFO [pool-1-thread-1] ShutdownHookManager: Deleting directory /tmp/spark-72515c56-2ba4-4b36-942c-xxx

Verificar resultados

1. Faça login no console do Kibana do seu cluster Elasticsearch. Consulte Log on to the Kibana console.

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

3. Na aba Console, execute a seguinte consulta para confirme que os dados foram gravados com sucesso:

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

   Uma resposta bem-sucedida retorna os três documentos gravados pelo job do Spark:

   GET company/_search
   {
     "query": {
       "match_all": {}
     }
   }
   
   --- Response ---
   
   {
     "took" : 0,
     "timed_out" : false,
     "_shards" : {
       "total" : 1,
       "successful" : 1,
       "skipped" : 0,
       "failed" : 0
     },
     "hits" : {
       "total" : 3,
       "max_score" : 1.0,
       "hits" : [
         {
           "_index" : "company",
           "_type" : "_doc",
           "_id" : "1MDGq3UBNM177Uedd0uo",
           "_score" : 1.0,
           "_source" : {
             "employees0" : "{"id": 1, "name": "zhangsan", "birth": "1990-01-01", "addr": "No.969, wenyixi Rd, yuhang, hangzhou"}"
           }
         },
         {
           "_index" : "company",
           "_type" : "_doc",
           "_id" : "1cDGq3UBNM177Uedd0uo",
           "_score" : 1.0,
           "_source" : {
             "employees1" : "{"id": 2, "name": "lisi", "birth": "1991-01-01", "addr": "No.556, xixi Rd, xihu, hangzhou"}"
           }
         },
         {
           "_index" : "company",
           "_type" : "_doc",
           "_id" : "1sDGq3UBNM177Uedd0uo",
           "_score" : 1.0,
           "_source" : {
             "employees2" : "{"id": 3, "name": "wangwu", "birth": "1992-01-01", "addr": "No.699 wangshang Rd, binjiang, hangzhou"}"
           }
         }
       ]
     }
   }

Próximos passos

O ES-Hadoop oferece suporte a muito mais do que apenas gravações Java RDD. Após a integração com o Spark, você também pode usar Spark Datasets, Spark Streaming, Scala e Spark SQL. Para mais detalhes, consulte o suporte ao Apache Spark.