Hadoop 3.1.4实战：DBInputFormat/DBOutputFormat实现MySQL与HDFS高效数据交换

在数据处理领域，ETL工程师经常面临一个经典难题：如何高效地在关系型数据库（如MySQL）与Hadoop分布式文件系统（HDFS）之间迁移海量数据。传统做法需要先将数据导出为中间文件，再通过HDFS命令上传，这种"导出-上传"的两步操作不仅效率低下，还容易因网络波动或存储限制导致失败。本文将深入解析Hadoop 3.1.4提供的DBInputFormat和DBOutputFormat组件，通过Java代码实战演示如何建立数据库与HDFS的直连通道。

1. 技术方案选型与环境准备

1.1 传统方案 vs 直连方案对比

传统ETL流程的三大瓶颈：

• 磁盘I/O瓶颈：数据需要先落地为临时文件
• 网络传输开销：文件需要二次传输到HDFS
• 处理延迟：多步骤操作增加整体耗时

DBInputFormat/DBOutputFormat优势：

• 零中间文件：直接从数据库读取/写入
• 并行分片：支持按条件分片读取（如ID范围）
• 事务批处理：通过批量提交提升写入效率

1.2 必备组件清单

组件	版本要求	作用
Hadoop	≥3.1.4	提供MapReduce框架
MySQL Connector/J	≥5.1.46	JDBC驱动
Maven	≥3.6.0	依赖管理

核心Maven依赖配置：

<dependency>
  <groupId>org.apache.hadoop</groupId>
  <artifactId>hadoop-client</artifactId>
  <version>3.1.4</version>
</dependency>
<dependency>
  <groupId>mysql</groupId>
  <artifactId>mysql-connector-java</artifactId>
  <version>5.1.46</version>
</dependency>

2. 从MySQL到HDFS：DBInputFormat实战

2.1 数据模型设计

实现DBWritable接口的User类示例：

public class User implements Writable, DBWritable {
    private int id;
    private String userName;
    // 其他字段...
    
    @Override
    public void write(PreparedStatement ps) throws SQLException {
        ps.setInt(1, id);
        ps.setString(2, userName);
        // 其他字段绑定...
    }
    
    @Override
    public void readFields(ResultSet rs) throws SQLException {
        this.id = rs.getInt("id");
        this.userName = rs.getString("user_name");
        // 其他字段读取...
    }
}

2.2 分片读取优化策略

通过DBInputFormat.setInput()配置分片参数：

// 按ID范围分片（适合有自增主键的表）
DBInputFormat.setInput(job, User.class,
    "SELECT id, user_name FROM users WHERE id >= ? AND id <= ?",
    "SELECT MIN(id), MAX(id) FROM users");

分片参数调优建议：

• 每个分片建议处理100-500万条记录
• 避免产生过多分片导致任务调度开销
• 复杂查询可改用inputQuery+inputCountQuery组合

2.3 数据一致性验证

通过MapReduce计数器实现记录数校验：

public class MySQLMapper extends Mapper<LongWritable, User, ...> {
    private Counter recordCounter;
    
    protected void setup(Context context) {
        recordCounter = context.getCounter("DATA_QUALITY", "INPUT_RECORDS");
    }
    
    protected void map(LongWritable key, User value, Context context) {
        recordCounter.increment(1);
        // 处理逻辑...
    }
}

运行后可在日志中查看：

DATA_QUALITY
    INPUT_RECORDS=12606948

3. 从HDFS到MySQL：DBOutputFormat实战

3.1 批量写入优化

关键配置参数：

// 设置每批次提交记录数
conf.setInt("mapreduce.jdbc.batch.size", 5000);

// 启用事务批量提交
conf.setBoolean("mapreduce.jdbc.transactional", true);

性能对比测试数据：

批量大小	写入100万条耗时(s)	数据库负载(%)
1	423	85
1000	217	62
5000	189	58
10000	175	55

3.2 异常处理机制

实现重试逻辑的Reducer示例：

public class MySQLReducer extends Reducer<...> {
    private static final int MAX_RETRY = 3;
    
    protected void reduce(...) {
        int retryCount = 0;
        while (retryCount < MAX_RETRY) {
            try {
                context.write(value, null);
                break;
            } catch (Exception e) {
                retryCount++;
                Thread.sleep(1000 * retryCount);
            }
        }
    }
}

4. 生产环境调优指南

4.1 连接池配置

在mapred-site.xml中添加：

<property>
  <name>mapreduce.jdbc.connection.max</name>
  <value>20</value>
</property>
<property>
  <name>mapreduce.jdbc.connection.timeout</name>
  <value>30000</value>
</property>

4.2 内存管理

常见OOM解决方案：

1. 增加Mapper堆内存：

   job.getConfiguration().set("mapreduce.map.memory.mb", "2048");
   

2. 减少ResultSet缓存行数：

   // 在查询中添加LIMIT子句
   DBInputFormat.setInput(job, User.class,
       "SELECT * FROM users LIMIT 1000000",
       "SELECT COUNT(*) FROM users");
   

4.3 监控指标

通过JMX暴露的关键指标：

• DBInputFormat.RecordsRead
• DBOutputFormat.BatchesCommitted
• ConnectionPool.ActiveConnections

5. 典型应用场景解析

5.1 用户画像数据同步

场景特点：

• 源表数据量：5000万+
• 字段包含JSON等复杂类型
• 需要增量同步

优化方案：

// 增量同步SQL示例
String incrementalSQL = "SELECT * FROM user_profiles " +
    "WHERE update_time > STR_TO_DATE(?, '%Y-%m-%d %H:%i:%s')";

DBInputFormat.setInput(job, UserProfile.class,
    incrementalSQL,
    "SELECT MAX(update_time) FROM user_profiles");

5.2 电商订单分析

挑战：

• 订单表存在多表关联
• 需要处理事务一致性

解决方案：

1. 创建数据库视图：

   CREATE VIEW order_analysis_view AS
   SELECT o.*, u.user_name 
   FROM orders o JOIN users u ON o.user_id = u.id;
   

2. 直接读取视图：

   DBInputFormat.setInput(job, Order.class,
       "SELECT * FROM order_analysis_view",
       "SELECT COUNT(*) FROM order_analysis_view");
   

通过本文的实战演示可以看到，合理运用DBInputFormat和DBOutputFormat能够将传统需要数小时的ETL过程缩短到分钟级。特别是在处理千万级以上的数据迁移任务时，这种直连方式相比传统文件交换方案有着明显的性能优势。