elasticsearch-hadoop源码剖析：从输入格式到输出格式的完整实现

【免费下载链接】elasticsearch-hadoop :elephant: Elasticsearch real-time search and analytics natively integrated with Hadoop 项目地址: https://gitcode.com/gh_mirrors/el/elasticsearch-hadoop

elasticsearch-hadoop是一个实现Elasticsearch与Hadoop生态系统深度集成的开源项目，它提供了高效的数据读写能力，使Hadoop作业能够直接与Elasticsearch进行交互。本文将深入剖析其核心组件——输入格式（InputFormat）和输出格式（OutputFormat）的实现原理，帮助开发者理解数据在Hadoop与Elasticsearch之间的流转过程。

核心组件概览：InputFormat与OutputFormat的角色

在Hadoop MapReduce框架中，InputFormat和OutputFormat是数据输入输出的关键接口。elasticsearch-hadoop通过实现这两个接口，分别提供了EsInputFormat和EsOutputFormat类，实现了从Elasticsearch读取数据和向Elasticsearch写入数据的功能。

• 输入流程：EsInputFormat负责将Elasticsearch中的数据分割为MapReduce可处理的输入分片（InputSplit），并通过RecordReader读取数据。
• 输出流程：EsOutputFormat负责将MapReduce的输出结果写入Elasticsearch，处理数据序列化和批量提交。

这两个组件的实现位于项目的mr模块中，具体路径为：

• EsInputFormat.java
• EsOutputFormat.java

EsInputFormat：从Elasticsearch读取数据的实现

1. 核心功能与类结构

EsInputFormat实现了Hadoop新旧API的InputFormat接口，其核心功能包括：

• 将Elasticsearch查询结果分割为多个InputSplit
• 创建RecordReader读取分片数据
• 支持两种输出格式：MapWritable（默认）和JSON文本

类结构上，EsInputFormat包含多个内部类协同工作：

• EsInputSplit：封装Elasticsearch分片信息
• EsInputRecordReader：抽象基类，实现数据读取逻辑
• WritableEsInputRecordReader：默认实现，输出MapWritable
• JsonWritableEsInputRecordReader：输出JSON文本

2. 分片策略与实现

在MapReduce中，InputSplit决定了数据如何分配给不同的Mapper。EsInputFormat的分片逻辑如下：

// 简化代码：获取Elasticsearch分片信息并创建InputSplit
Collection<PartitionDefinition> partitions = RestService.findPartitions(settings, log);
EsInputSplit[] splits = new EsInputSplit[partitions.size()];
int index = 0;
for (PartitionDefinition part : partitions) {
    splits[index++] = new EsInputSplit(part);
}

这段代码位于EsInputFormat.getSplits()方法中，通过RestService获取Elasticsearch集群的分片信息，每个分片对应一个EsInputSplit。这种设计确保了Hadoop的并行计算能力能够充分利用Elasticsearch的分布式架构。

3. 数据读取流程

RecordReader是实际读取数据的组件，EsInputRecordReader的工作流程如下：

1. 初始化：通过RestService.createReader()建立与Elasticsearch的连接，创建ScrollQuery用于高效批量查询
2. 数据读取：通过scrollQuery.next()获取文档数据，转换为Hadoop Writable格式
3. 资源释放：作业完成后关闭ScrollQuery和RestRepository，释放连接资源

关键代码实现：

// 数据读取核心逻辑
boolean hasNext = scrollQuery.hasNext();
if (!hasNext) {
    return false;
}
Object[] next = scrollQuery.next();
currentKey = setCurrentKey(key, next[0]);  // 文档ID
currentValue = setCurrentValue(value, next[1]);  // 文档内容

EsOutputFormat：向Elasticsearch写入数据的实现

1. 核心功能与批量写入

EsOutputFormat负责将MapReduce的输出结果写入Elasticsearch，其核心特点是：

• 支持Hadoop新旧API
• 使用批量写入（Bulk API）提高性能
• 处理数据序列化和错误重试

与输入格式类似，EsOutputFormat也通过内部类EsRecordWriter实现具体的写入逻辑。

2. 写入流程与事务处理

数据写入的核心流程在EsRecordWriter.write()方法中实现：

@Override
public void write(Object key, Object value) throws IOException {
    if (!initialized) {
        initialized = true;
        init();  // 初始化连接和资源
    }
    repository.writeToIndex(value);  // 写入数据
}

RestRepository负责管理与Elasticsearch的连接和批量写入操作。为了确保数据可靠性，EsOutputFormat还实现了OutputCommitter接口，但由于Elasticsearch本身不支持Hadoop的事务机制，这里主要做了空实现以满足Hadoop框架要求。

3. 配置检查与优化建议

在作业提交前，EsOutputFormat会进行必要的配置检查：

// 检查输出配置
Assert.hasText(settings.getResourceWrite(), "No resource specified");
InitializationUtils.discoverClusterInfo(settings, log);
InitializationUtils.checkIndexExistence(settings);

同时，针对Hadoop的推测执行机制，EsOutputFormat会给出警告：

if (HadoopCfgUtils.getSpeculativeMap(cfg)) {
    log.warn("Speculative execution enabled for mapper - consider disabling it to prevent data corruption");
}

这是因为推测执行可能导致重复写入数据，建议在写入Elasticsearch时禁用该功能。

多框架集成：Hive与Spark的适配

elasticsearch-hadoop不仅支持MapReduce，还为Hive和Spark提供了专用的输入输出格式实现。

Hive集成

在Hive模块中，EsHiveInputFormat和EsHiveOutputFormat继承自核心的EsInputFormat和EsOutputFormat，并针对Hive的特性进行了适配：

// Hive输入格式实现
public class EsHiveInputFormat extends EsInputFormat<Text, Writable> {
    // 适配Hive的序列化和类型系统
}

// Hive存储处理器
public class EsStorageHandler implements StorageHandler {
    @Override
    public Class<? extends InputFormat> getInputFormatClass() {
        return EsHiveInputFormat.class;
    }
    @Override
    public Class<? extends OutputFormat> getOutputFormatClass() {
        return EsHiveOutputFormat.class;
    }
}

相关代码位于：hive/src/main/java/org/elasticsearch/hadoop/hive/

Spark集成

Spark通过RDD API使用elasticsearch-hadoop，在spark/core模块中提供了对EsInputFormat的封装：

// Spark RDD创建示例
JavaPairRDD data = sc.hadoopRDD(hdpConf, EsInputFormat.class, NullWritable.class, MapWritable.class);

Spark SQL还提供了DataFrame API，进一步简化了与Elasticsearch的交互。相关实现位于：spark/core/src/main/scala/org/elasticsearch/spark/rdd/

实际应用示例

MapReduce作业配置

使用EsInputFormat和EsOutputFormat的典型MapReduce作业配置如下：

// 读取Elasticsearch
job.setInputFormatClass(EsInputFormat.class);
EsInputFormat.setInput(job.getConfiguration(), "index_name/type_name");
EsInputFormat.setQuery(job.getConfiguration(), "{\"query\": {\"match_all\": {}}}");

// 写入Elasticsearch
job.setOutputFormatClass(EsOutputFormat.class);
EsOutputFormat.setOutput(job.getConfiguration(), "output_index/type_name");

关键配置参数

elasticsearch-hadoop提供了丰富的配置参数，控制数据读写行为：

• es.resource：指定Elasticsearch索引和类型
• es.query：查询语句，用于筛选数据
• es.nodes：Elasticsearch节点地址
• es.batch.size.bytes：批量写入大小限制

完整的配置说明可参考项目文档：docs/reference/configuration.md

总结与最佳实践

elasticsearch-hadoop通过EsInputFormat和EsOutputFormat实现了Hadoop与Elasticsearch的无缝集成，其设计充分考虑了分布式计算的特点和性能优化。在实际应用中，建议：

1. 合理设置分片：根据Elasticsearch集群规模调整Map任务数量
2. 优化批量参数：调整es.batch.size等参数，平衡网络传输和内存占用
3. 禁用推测执行：避免数据重复写入
4. 监控性能指标：通过Hadoop计数器和Elasticsearch监控API跟踪作业状态

通过深入理解这些核心组件的实现原理，开发者可以更好地利用elasticsearch-hadoop的能力，构建高效的大数据处理管道。项目的完整源码和更多细节可参考：mr/src/main/java/org/elasticsearch/hadoop/mr/

【免费下载链接】elasticsearch-hadoop :elephant: Elasticsearch real-time search and analytics natively integrated with Hadoop 项目地址: https://gitcode.com/gh_mirrors/el/elasticsearch-hadoop