Flink中HadoopInputFormat作为StreamExecutionEnvironment输入能否使用批模式？

调用StreamExecutionEnvironment.createInput配合HadoopInputFormat时可以使用批模式，报错是因为默认将该源识别为无界流导致的。

报错原因 #

Flink 1.13 版本中，直接调用StreamExecutionEnvironment.createInput方法创建基于HadoopInputFormat的数据源时，默认会将该数据源标记为UNBOUNDED（无界流），而Flink的批执行模式明确禁止无界源输入，因此触发了你遇到的异常。

可行实现方案 #

提供两种可选方案，可根据技术栈选择：

方案1：使用批处理专用环境ExecutionEnvironment（旧API兼容方案） #

如果不需要用到流批一体的新特性，直接用Flink原生的批处理环境即可，天然识别HadoopInputFormat为有界源，不需要额外配置：

// 初始化批处理专用环境
ExecutionEnvironment batchEnv = ExecutionEnvironment.getExecutionEnvironment();
// 原有构造HBase快照输入的逻辑保持不变
Job job = Job.getInstance(conf);
TableSnapshotInputFormat.setInput(job, snapshotName, new Path(restoreDir));
HadoopInputFormat<ImmutableBytesWritable, Result> input = HadoopInputs.createHadoopInput(
  new TableSnapshotInputFormat(), ImmutableBytesWritable.class, Result.class, job);
// 直接用批环境读取源
DataSource<Tuple2<ImmutableBytesWritable, Result>> hbaseDataSource = batchEnv.createInput(input);
// 后续编写批处理逻辑即可

方案2：流批一体API下显式指定源为有界（推荐方案） #

如果要继续使用StreamExecutionEnvironment的流批一体架构，只需要把原有createInput调用改为显式构造有界源即可：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 显式指定为批执行模式
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
// 原有构造HBase快照输入的逻辑保持不变
Job job = Job.getInstance(conf);
TableSnapshotInputFormat.setInput(job, snapshotName, new Path(restoreDir));
HadoopInputFormat<ImmutableBytesWritable, Result> hadoopInputFormat = HadoopInputs.createHadoopInput(
  new TableSnapshotInputFormat(), ImmutableBytesWritable.class, Result.class, job);
// 包装为有界源后构造DataStream
DataStream<Tuple2<ImmutableBytesWritable, Result>> hbaseSource = env
  .fromInputFormat(hadoopInputFormat)
  .bounded(Boundedness.BOUNDED)
  .build();
// 后续正常编写处理逻辑即可

注意事项 #

• 请确保项目中已经引入对应版本的flink-hadoop-compatibility依赖，否则会找不到相关类
• HBase客户端依赖需要和集群端版本兼容，避免出现快照读取失败的问题

内容的提问来源于stack exchange，提问作者陈兆卿