Hadoop Formats 在 Flink 里复用 Hadoop InputFormat（flink-hadoop-compatibility）

1、项目依赖配置

核心依赖是 Flink 的兼容模块：

<dependency><groupId>org.apache.flink</groupId><artifactId>flink-hadoop-compatibility</artifactId><version>2.2.0</version></dependency>

如果你要在本地（IDE / MiniCluster）跑起来，通常还需要带上一个 Hadoop 客户端依赖（本地有 class 才能初始化 Hadoop 相关类）：

<dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-client</artifactId><version>2.10.2</version><scope>provided</scope></dependency>

实战建议：

• 集群环境里 Hadoop 依赖往往由平台（YARN/HDFS）提供，你本地为了能跑通需要加hadoop-client；
• 如果你是打 fat jar 跑 standalone/K8s，就要更谨慎处理 Hadoop 依赖冲突（常见是guava、jackson、netty版本冲突）。

2、两种包法：readHadoopFile vs createHadoopInput

Flink 通过HadoopInputs提供两种方式把 Hadoop InputFormat 包装成 Flink InputFormat：

• readHadoopFile(...)：用于继承自 Hadoop 的 FileInputFormat的场景（典型：读 HDFS 文件类格式）
• createHadoopInput(...)：用于通用 InputFormat（不一定是文件类，例如某些外部系统的 InputFormat）

包装后的 Source 输出类型是一个二元组：Tuple2<K, V>

• f0：key
• f1：value

这点非常重要：你后续处理 DataStream 的时候，要把 key/value 拆开理解。

3、示例：使用 Hadoop 的 KeyValueTextInputFormat

下面示例演示如何把 Hadoop 的KeyValueTextInputFormat作为 Flink Source：

importorg.apache.flink.api.java.tuple.Tuple2;importorg.apache.flink.streaming.api.datastream.DataStream;importorg.apache.flink.streaming.api.environment.StreamExecutionEnvironment;importorg.apache.flink.hadoopcompatibility.HadoopInputs;importorg.apache.hadoop.io.Text;importorg.apache.hadoop.mapred.KeyValueTextInputFormat;publicclassHadoopInputFormatDemo{publicstaticvoidmain(String[]args)throwsException{StreamExecutionEnvironmentenv=StreamExecutionEnvironment.getExecutionEnvironment();StringtextPath="hdfs:///path/to/input";KeyValueTextInputFormattextInputFormat=newKeyValueTextInputFormat();DataStream<Tuple2<Text,Text>>input=env.createInput(HadoopInputs.readHadoopFile(textInputFormat,Text.class,Text.class,textPath));// 业务处理：把 Text 转成 StringDataStream<String>lines=input.map(kv->"key="+kv.f0.toString()+", value="+kv.f1.toString());lines.print();env.execute("hadoop-inputformat-demo");}}

4、典型使用场景与坑位

常见适配场景：

• 你要读一个历史遗留的 Hadoop InputFormat（比如某个自研存储、HBase/Hive 的某些格式、云厂商的 InputFormat）
• 已经有成熟的 MR InputFormat + 配置参数（账号、表名、列族、分区等），直接复用最省时间

容易踩的坑：

• 依赖冲突：Hadoop 生态依赖链很长，尤其本地 IDE 跑最容易爆 classpath 冲突
• Key/Value 语义：输出一定是Tuple2<K,V>，不要直接当成一行文本
• 配置注入：很多 InputFormat 依赖 Hadoop Configuration（账号、endpoint、table 等），你需要在创建 InputFormat 或 Job/Configuration 时把参数设全
• 批/流差异：很多 Hadoop InputFormat 本质是“批式读取”，更适合 BATCH runtime；如果你硬塞进 streaming job，它也只是周期性读一遍（不是增量监控）、