Use ES-Hadoop to write data from HDFS to Elasticsearch - Elasticsearch

Procedure

Preparations

Create an Alibaba Cloud Elasticsearch instance and an E-MapReduce (EMR) instance in the same virtual private cloud (VPC). Then, enable the automatic index creation feature for the Elasticsearch instance, and prepare test data and a Java environment.
Step 1: Upload the ES-Hadoop JAR package to HDFS

Download the ES-Hadoop installation package and upload it to the HDFS directory on the master node of the EMR cluster.
Step 2: Configure pom dependencies

Create a Java Maven project and configure its pom dependencies.
Step 3: Write and run the MapReduce task

Write the Java code for the MapReduce task to write data to Elasticsearch. Package the code into a JAR file, upload it to the EMR cluster, and run the code to complete the data writing task.
Step 4: Verify the results

In the Kibana console of your Elasticsearch instance, view the data written by the MapReduce task.

Preparations

Create an Alibaba Cloud Elasticsearch instance and enable the automatic index creation feature.

For more information, see Create an Alibaba Cloud Elasticsearch instance and Configure YML parameters. This topic uses an Elasticsearch 6.7.0 instance as an example.

Important
In a production environment, you must disable the automatic index creation feature. Create an index and its mapping in advance. Because this topic is for testing purposes only, the automatic index creation feature is enabled.
Create an EMR instance in the same VPC as the Elasticsearch instance.
Use the following instance configurations:
- Edition: EMR-3.29.0
- Required Services: HDFS 2.8.5. Keep the default settings for other services.
For more information, see Create a cluster.
Important
The private IP address whitelist for an Elasticsearch cluster is set to 0.0.0.0/0 by default. You can view this setting on the security configuration page. If you change this default, you must add the EMR cluster's internal IP address to the whitelist:

See View cluster list and details to obtain the internal IP address of the EMR cluster.

See Configure a public or private IP address whitelist for an Elasticsearch cluster to configure the VPC private IP address whitelist for the Elasticsearch cluster.

Prepare JSON test data, write it to the map.json file, and upload the file to the /tmp/hadoop-es directory in HDFS.

This topic uses the following test data.

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

Prepare a Java environment. The JDK version must be 1.8.0 or later.

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

Download the ES-Hadoop installation package whose version matches your Elasticsearch cluster.

This topic uses elasticsearch-hadoop-6.7.0.zip.
Log on to the EMR consol, obtain the IP address of the master node, and then use SSH to log on to the corresponding ECS instance.

For more information, see Log on to a cluster.
Upload the elasticsearch-hadoop-6.7.0.zip package to the master node and decompress the package to obtain the elasticsearch-hadoop-6.7.0.jar file.

Create an HDFS directory and upload the elasticsearch-hadoop-6.7.0.jar file to the directory.

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

Step 2: Configure pom dependencies

Create a Java Maven project and add the following pom dependencies to the pom.xml file of the project.

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

Important

Ensure that the versions in the pom dependencies match the corresponding service versions. For example, the elasticsearch-hadoop-mr version must match the Alibaba Cloud Elasticsearch version, and the hadoop-hdfs version must match the HDFS version.

Step 3: Write and run the MapReduce task

Write the sample code.

The following code reads the JSON files from the /tmp/hadoop-es directory in HDFS. It then writes each line from these files as a document to Elasticsearch. The EsOutputFormat class completes the write operation in the Map phase.

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);
    }
}

Table 1. ES-Hadoop parameter descriptions

Parameter	Default value	Description
es.nodes	localhost	The endpoint of the Alibaba Cloud Elasticsearch instance. Use the private endpoint. You can find it on the Basic Information page of the instance. For more information, see View the basic information of an instance.
es.port	9200	The port number of the Elasticsearch instance.
es.net.http.auth.user	elastic	The username to access the Elasticsearch instance. Note If you specify the `elastic` account in your application, any subsequent password change for this account can cause temporary service disruptions due to propagation delay. Therefore, using the `elastic` account is not recommended. Instead, create a dedicated user with appropriate permissions in the Kibana console. For more information, see Use the RBAC mechanism of Elasticsearch X-Pack to control user access.
es.net.http.auth.pass	/	The password to access the Elasticsearch instance.
es.nodes.wan.only	false	Specifies whether to perform node sniffing when you connect to an Elasticsearch cluster that uses a virtual IP address in the cloud. true: Enabled false: The parameter is not set.
es.nodes.discovery	true	Specifies whether to disable node discovery. true: Disable false: Not disabled
es.input.use.sliced.partitions	true	Specifies whether to use slice partitions. true: Uses slice partitions. Setting this parameter to `true` can significantly increase the index pre-read time, sometimes making it much longer than the query time itself. We recommend that you set this parameter to `false` to improve query performance. false: Does not use slice partitions.
es.index.auto.create	true	Specifies whether to automatically create an index if it does not exist when you write data from a Hadoop component to an Elasticsearch cluster. true: Automatically create the index. false: Automatic creation is disabled.
es.resource	/	The index and type to read from or write to.
es.input.json	false	Specifies whether the input is in JSON format. true: The input is in JSON format. false: The input is not in JSON format.
es.mapping.names	/	The mapping between table fields and Elasticsearch index fields.
es.read.metadata	false	Enable this property if your operation involves internal Elasticsearch fields, such as _id.

For more information about ES-Hadoop configuration items, see the official configuration documentation.

Package the code into a JAR file and upload it to an EMR client machine, such as a gateway or the master node of the EMR cluster.
On the EMR client machine, run the following command to execute the MapReduce program.
```
hadoop jar es-mapreduce-1.0-SNAPSHOT.jar /tmp/hadoop-es/map.json
```
Note
Replace es-mapreduce-1.0-SNAPSHOT.jar with the name of your uploaded JAR file.

Step 4: Verify the results

Log in to the Kibana console for your Alibaba Cloud Elasticsearch instance.

For more information, see Log in to the Kibana console.
In the left navigation pane, click Dev Tools.

On the Console tab, run the following command to view the data written by the MapReduce task.

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

A successful query returns the following result.

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

Summary

This topic demonstrates how to use ES-Hadoop to write data to Elasticsearch with a MapReduce task, using Alibaba Cloud Elasticsearch and EMR as an example. You can also use a MapReduce task to query data from Elasticsearch. The configuration for querying is similar to that for writing. For more information, see the official documentation for Reading data from Elasticsearch.