Flink自定义InputFormat中如何正确获取RuntimeContext实例？

你碰到的IllegalStateException本质原因很简单：普通的DelimitedInputFormat并不属于Flink的「富函数（Rich Functions）」体系，Flink运行时不会给它注入RuntimeContext实例，自然调用getRuntimeContext()就会报错。

要解决这个问题，你需要把自定义输入格式改成继承RichInputFormat——这是Flink专门为需要访问运行时上下文（比如累加器、配置、状态等）的输入格式提供的基类。

具体实现方案 #

如果你想复用DelimitedInputFormat的分隔符解析逻辑，可以通过委托模式来实现：让你的自定义类继承RichInputFormat，内部持有一个DelimitedInputFormat实例，把大部分方法委托给它处理，同时在富函数的生命周期方法里获取RuntimeContext并初始化累加器。

示例代码如下（Scala）：

import org.apache.flink.api.common.io.{RichInputFormat, DelimitedInputFormat, FileInputSplit}
import org.apache.flink.configuration.Configuration
import org.apache.flink.api.common.accumulators.IntCounter

class MyInputFormat[T](delimiter: Array[Byte], serializer: (Array[Byte]) => T) 
  extends RichInputFormat[T, FileInputSplit] {

  @transient private var delimitedFormat: DelimitedInputFormat[T] = _
  @transient var lineCounter: IntCounter = _

  // 配置DelimitedInputFormat的核心参数
  override def configure(parameters: Configuration): Unit = {
    delimitedFormat = new DelimitedInputFormat[T](null) // 路径可后续在外部设置，或直接传入
    delimitedFormat.setDelimiter(delimiter)
    // 设置自定义反序列化逻辑，将字节数组转为目标类型T
    delimitedFormat.setRecordConverter(serializer)
    delimitedFormat.configure(parameters)
  }

  // 这里是获取RuntimeContext的正确时机：每个任务实例初始化时仅调用一次
  override def openInputFormat(): Unit = {
    lineCounter = new IntCounter()
    getRuntimeContext.addAccumulator("rowsInFile", lineCounter)
  }

  // 打开分片时，委托给DelimitedInputFormat处理
  override def open(split: FileInputSplit): Unit = {
    delimitedFormat.open(split)
  }

  // 读取下一条记录时，同步更新累加器
  override def nextRecord(reuse: T): T = {
    val record = delimitedFormat.nextRecord(reuse)
    if (record != null) {
      lineCounter.add(1)
    }
    record
  }

  // 其他方法全部委托给DelimitedInputFormat
  override def close(): Unit = delimitedFormat.close()
  override def reachedEnd(): Boolean = delimitedFormat.reachedEnd()
  override def createInputSplits(minNumSplits: Int): Array[FileInputSplit] = 
    delimitedFormat.createInputSplits(minNumSplits)
  override def getInputSplitAssigner(splits: Array[FileInputSplit]): InputSplitAssigner = 
    delimitedFormat.getInputSplitAssigner(splits)
}

关键注意点 #

1. 使用openInputFormat而非open初始化累加器：open方法会为每个输入分片调用一次，而openInputFormat是每个任务实例仅调用一次，累加器绑定到任务实例，放在这里可避免重复添加累加器。
2. 委托模式复用原有逻辑：直接继承RichInputFormat后，无需重新实现分隔符解析逻辑，只需把DelimitedInputFormat作为内部组件复用即可。
3. 序列化器的传递：根据你的目标类型T，需要实现字节数组到T的转换逻辑，在configure方法里通过setRecordConverter传入。

这样修改后，你的自定义输入格式就能正常获取RuntimeContext并使用累加器了，不会再抛出IllegalStateException。

内容的提问来源于stack exchange，提问作者radumanolescu