小菜鸟

Hadoop 数据写入 Elasticsearch：基于 ES-Hadoop 的实现详解

ES-Hadoop 是 Elasticsearch 官方推出的工具，用于打通 Hadoop 生态（HDFS、MapReduce、Spark 等）与 Elasticsearch，支持数据双向流动。本文详细讲解如何通过 MapReduce 任务将 Hadoop 中的数据（如 HDFS 上的文件）写入 Elasticsearch，包括依赖配置、代码实现、核心参数及最佳实践。

ES-Hadoop 简介与依赖准备

核心作用

ES-Hadoop 简化了 Hadoop 与 Elasticsearch 的数据交互，提供：

• 适配 Hadoop 生态的输入 / 输出格式（EsInputFormat/EsOutputFormat）。
• 自动处理数据序列化（Hadoop 数据 → ES 文档 JSON）。
• 支持批量写入、节点发现、认证等核心功能。

依赖配置

在 Maven 项目中添加以下依赖（版本需与 Hadoop 和 Elasticsearch 匹配）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

<!-- Hadoop 核心依赖 -->
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-common</artifactId>
    <version>2.7.3</version> <!-- 与集群 Hadoop 版本一致 -->
    <exclusions>
        <exclusion> <!-- 排除冲突的日志组件 -->
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
        </exclusion>
    </exclusions>
</dependency>
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-client</artifactId>
    <version>2.7.3</version>
</dependency>

<!-- ES-Hadoop 核心依赖 -->
<dependency>
    <groupId>org.elasticsearch</groupId>
    <artifactId>elasticsearch-hadoop</artifactId>
    <version>6.8.23</version> <!-- 与 ES 版本一致 -->
</dependency>

<!-- 可选：ES 高级客户端（用于额外操作） -->
<dependency>
    <groupId>org.elasticsearch.client</groupId>
    <artifactId>elasticsearch-rest-high-level-client</artifactId>
    <version>6.8.23</version>
</dependency>

MapReduce 任务实现：HDFS 数据写入 ES

核心思路

通过 MapReduce 任务读取 HDFS 上的文件（如 JSON 格式），经 Mapper 处理后，通过 EsOutputFormat 批量写入 Elasticsearch。由于仅需数据透传（无聚合逻辑），无需 Reducer。

代码实现

（1）Job 配置类（H2EJob.java）

Elasticsearch 索引监控：全方位追踪索引状态与性能

索引是 Elasticsearch 存储和查询数据的核心单元，其健康状态和性能直接影响整个集群的可用性。Elasticsearch 提供了一系列 API 用于监控索引的统计信息、分片状态、恢复进度等关键指标，帮助开发者和运维人员及时发现并解决问题。

索引统计信息（_stats）

_stats API 提供索引的详细统计数据，涵盖文档数量、存储大小、索引 / 查询性能、缓存使用等维度，是监控索引整体状态的核心工具。

基本用法

• 查看所有索引的统计信息：

  1	GET _stats // 返回集群中所有索引的统计数据

• 查看指定索引的统计信息：

  1	GET my_index/_stats // 仅返回my_index的统计数据

• 过滤统计维度（如仅查看存储和事务日志）：

  1	GET my_index/_stats/store,translog // 仅返回存储和事务日志信息

核心统计指标

（1）文档与存储信息（docs + store）

1
2
3
4
5
6
7
8
9
10
11

{
  "primaries": {
    "docs": {
      "count": 10000,    // 文档总数
      "deleted": 500     // 标记删除的文档数（未实际删除，等待合并）
    },
    "store": {
      "size_in_bytes": 52428800  // 索引占用磁盘空间（50MB）
    }
  }
}

Maven 使用外部依赖：本地 Jar 包的引入方案

在开发中，经常会遇到需要使用中央仓库或远程私服中不存在的外部 Jar 包（如第三方非开源 SDK、定制化组件）的情况。Maven 提供了多种方式引入这类本地依赖，本文将详细介绍最常用的两种方案，并分析其优缺点。

使用 system 范围引入（简单直接）

system 是 Maven 依赖范围的一种，专门用于引用本地文件系统中的 Jar 包，无需通过仓库下载。

配置方式

1
2
3
4
5
6
7
8
9
10

<dependencies>
  <dependency>
    <groupId>com.example</groupId> <!-- 自定义组织标识 -->
    <artifactId>custom-sdk</artifactId> <!-- 自定义项目标识 -->
    <version>1.0.0</version> <!-- 自定义版本号 -->
    <scope>system</scope> <!-- 声明为 system 范围 -->
    <!-- 本地 Jar 包的路径（支持绝对路径或相对路径） -->
    <systemPath>${project.basedir}/src/lib/custom-sdk.jar</systemPath>
  </dependency>
</dependencies>

• systemPath 路径说明：
  • ${project.basedir} 表示项目根目录（含 pom.xml 的目录），推荐使用相对路径（如 src/lib/），便于项目迁移。
  • 也可使用绝对路径（如 D:/libs/custom-sdk.jar），但会导致项目在其他机器上不可用，不推荐。

优缺点分析

安装到本地仓库（推荐方案）

将本地 Jar 包安装到 Maven 本地仓库（~/.m2/repository），使其像普通依赖一样被 Maven 管理，解决 system 范围的移植性问题。

操作步骤

1. 执行安装命令：
   使用 mvn install:install-file 命令将 Jar 包安装到本地仓库：

Elasticsearch 分片：分布式存储与查询的核心

分片（Shard）是 Elasticsearch 实现分布式存储和并行查询的基础，将一个索引的数据拆分为多个片段，分布在不同节点上。理解分片的工作原理和配置策略，对优化集群性能和可靠性至关重要。

分片的基本概念

主分片与副本分片

• 主分片（Primary Shard）：
  • 索引数据的原始存储单元，负责处理写入请求（索引、更新、删除）。
  • 数量在索引创建时指定（number_of_shards），创建后不可修改（因路由依赖主分片数量）。
  • 默认值：5 个（6.8.x 版本）。
• 副本分片（Replica Shard）：
  • 主分片的副本，用于冗余备份和分担查询压力（只读，不处理写入）。
  • 数量可动态调整（number_of_replicas），默认 1 个。
  • 副本分片不能与对应的主分片位于同一节点（避免单点故障）。

分片的核心作用

• 分布式存储：突破单节点存储上限，支持海量数据（如 10 个主分片可存储 10 倍于单节点的数据）。
• 并行处理：查询请求可同时在多个分片（主或副本）上执行，提升查询效率。
• 高可用性：主分片故障时，副本分片可自动升级为主分片（需集群有足够节点）。

分片路由机制

当文档写入或查询时，Elasticsearch 需要确定文档属于哪个主分片，这一过程通过路由算法实现：

路由公式

Elasticsearch 集群：分布式架构与管理详解

Elasticsearch 集群是由多个节点（Node）组成的分布式系统，通过协同工作实现高可用、高扩展的数据存储与查询。集群的核心是节点角色分工、分片（Shard）分布式管理及动态配置，以下从架构、监控、配置等方面详细解析。

集群核心组件

节点（Node）：集群的基本单元

节点是 Elasticsearch 的运行实例，每个节点通过 cluster.name 标识所属集群。根据功能分工，节点分为以下类型：

分片（Shard）：数据分布式存储的核心

分片是索引数据的最小存储单元，分为主分片和副本分片，实现数据分布式存储与高可用：

• 主分片（Primary Shard）：
  • 负责数据写入，数量在索引创建时指定（number_of_shards），不可动态修改。
  • 作用：通过横向扩展（增加主分片）提升存储容量。
• 副本分片（Replica Shard）：
  • 主分片的冗余副本，负责分担查询压力，数量可动态调整（number_of_replicas）。
  • 作用：提高查询吞吐量，保障数据可用性（主分片故障时自动升级为新主分片）。

集群监控与管理 API

Elasticsearch 提供丰富的 API 监控集群状态、节点信息及分片分布，核心接口如下：

集群状态监控

（1）查看集群健康状态