【原创】MapReduce备份Elasticsearch数据到HDFS(JAVA)

一、环境：JAVA8，Elasticsearch-5.6.2，Hadoop-2.8.1
二、实现功能：mapreduce读elasticsearch数据、输出parquet文件、多输出路径
三、主要依赖

<dependency>
    <groupId>org.elasticsearch.client</groupId>
    <artifactId>transport</artifactId>
    <version></version>
</dependency>
<dependency>
    <groupId>org.apache.logging.log4j</groupId>
    <artifactId>log4j-to-slf4j</artifactId>
    <version></version>
</dependency>
<dependency>
    <groupId>ch.qos.logback</groupId>
    <artifactId>logback-classic</artifactId>
    <version></version>
</dependency>
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-client</artifactId>
    <version></version>
</dependency>
<dependency>
    <groupId>org.apache.parquet</groupId>
    <artifactId>parquet-avro</artifactId>
    <version></version>
</dependency>
<dependency>
    <groupId>org.elasticsearch</groupId>
    <artifactId>elasticsearch-hadoop-mr</artifactId>
    <version></version>
</dependency>
<dependency>
    <groupId>org.apache.avro</groupId>
    <artifactId>avro-mapred</artifactId>
    <version></version>
</dependency>

四、主要代码
1.public class Job extends Configured implements Tool

Configuration conf = getConf();
conf.set(ConfigurationOptions.ES_NODES,"127.0.0.1");
conf.set(ConfigurationOptions.ES_PORT,"9200");
conf.set(ConfigurationOptions.ES_RESOURCE, "index/type");
conf.set(ConfigurationOptions.ES_QUERY, "?q=*");
Job job = Job.getInstance(conf);
// ...（其他不重要的设置）
// set input
job.setInputFormatClass(EsInputFormat.class);
// set output
job.setOutputFormatClass(AvroParquetOutputFormat.class);
AvroParquetOutputFormat.setOutputPath(job, ${outputDir});
AvroParquetOutputFormat.setSchema(job, ${schema});
AvroParquetOutputFormat.setCompression(job, CompressionCodecName.SNAPPY);
AvroParquetOutputFormat.setCompressOutput(job, true);
AvroParquetOutputFormat.setBlockSize(job, ${size});
for(String name: ${list}){
　　MultipleOutputs.addNamedOutput(job, name, AvroParquetOutputFormat.class, Void.class, GenericRecord.class);
}

2.public class Mapper extends Mapper<Text, MapWritable, ${KeyType}, ${ValueType}>

// 代码一般，略

3.public class Reducer extends Reducer<${KeyType}, ${ValueType}, Void, GenericRecord>

private MultipleOutputs<Void, GenericRecord> multipleOutputs;

@Override
protected void setup(Context context) throws IOException, InterruptedException {
    multipleOutputs = new MultipleOutputs<>(context);
}

@Override
public void reduce(${KeyType} key, Iterable<${ValueType}> values, Context context) throws IOException, InterruptedException {
    for(${ValueType} value:values){
    　　GenericData.Record avroRecord = new GenericData.Record(ReflectData.get().getSchema(${实体类}.class));// value转实体类
        avroRecord.put(${字段名}, ${字段值});
    　　// ... n多字段   　　　　　　　　
        multipleOutputs.write(${Job中的name}, null, avroRecord, ${输出hdfs的绝对路径});
    }
}

@Override
protected void cleanup(Context context) throws IOException, InterruptedException {
    multipleOutputs.close();
}

五、遇到的问题
1.查询字符串scroll失败
ConfigurationOptions.ES_QUERY，不需要urlEncode，否则反而会解析失败
例如查询带时间范围：?q=event_time:>=1509465600 AND event_time:<1512057600

2.多输出路径重复跑job，根路径冲突
Job中的输出路径不能存在否则会抛异常“org.apache.hadoop.mapred.FileAlreadyExistsException”，所以在创建Job时需要判断输出路径是否存在，存在则删除。
当时用MultipleOutputs时，Job中的${outputDir}和Reducer中的${输出hdfs的绝对路径}可以是完全不同的目录，Job中的输出路径会保存_matadata等不是很重要的数据（parquet本身包含这些信息），Reducer中的输出路径为想要的输出路径，路径下只保存parquet文件。
重复执行相同的Job时删除Job中的输出路径，主要数据没有影响，另外如果Reducer的输出路径有冲突可以在Job中循环删除。