1、UDF 四种写法一览

1.1 实现接口（最基础、最直观）

class MyMapFunction implements MapFunction<String, Integer> {
  @Override
  public Integer map(String value) { return Integer.parseInt(value); }
}
data.map(new MyMapFunction());

适用：需要复用、逻辑中等复杂的函数。

1.2 匿名类（就地定义，便于临时逻辑）

data.map(new MapFunction<String, Integer>() {
  @Override
  public Integer map(String value) { return Integer.parseInt(value); }
});

适用：一次性、短小逻辑；可读性一般，慎滥用。

1.3 Java 8 Lambdas（最简洁）

data.filter(s -> s.startsWith("http://"));
data.reduce((i1, i2) -> i1 + i2);

适用：表达式风格、参数与返回类型清晰的简单逻辑。

提示：复杂逻辑或需要生命周期方法（如 open/close）时，请优先用 Rich Function。

1.4 Rich Function（增强版 UDF）

class MyMapFunction extends RichMapFunction<String, Integer> {
  @Override
  public Integer map(String value) { return Integer.parseInt(value); }
}
data.map(new MyMapFunction());

核心优势：

• 拥有 生命周期方法：open() / close()（适合资源初始化/释放）
• 可通过 getRuntimeContext() 获取 并发度、子任务索引、累加器注册 等能力
• 更易做 指标埋点、广播变量、外部资源连接池 等工程化工作

2、何时选择 Rich Function：生命周期与上下文能力

生命周期

• open(Configuration)：算子实例启动时调用。适合初始化连接池、注册累加器、读取配置。
• map/flatMap/filter/...：处理数据主流程。
• close()：算子实例退出时调用。适合释放资源、flush 缓冲。

RuntimeContext 常见用途

• getIndexOfThisSubtask()：分片调试与日志定位
• getNumberOfParallelSubtasks()：按并发规模分配资源
• addAccumulator(name, acc)：注册累加器（下节详解）

3、累加器与计数器：调试与数据洞察利器

累加器（Accumulator）：提供 add() 操作，作业结束后在 JobExecutionResult 中返回最终结果。
用途：统计处理行数、异常条数、分布直方图等，尤其适合离线/批处理或调试回放。

内置累加器

• IntCounter / LongCounter / DoubleCounter：数值累加
• Histogram：离散分箱统计（本质 Map<Integer, Integer>）

使用步骤（以计数行数为例）

public static class CountLinesMap extends RichMapFunction<String, String> {
  private final IntCounter numLines = new IntCounter();

  @Override
  public void open(Configuration parameters) {
    getRuntimeContext().addAccumulator("num-lines", numLines);
  }

  @Override
  public String map(String value) {
    numLines.add(1);
    return value;
  }
}

// 提交并等待完成（同步）
JobExecutionResult res = env.execute("acc-demo");
Integer lines = res.getAccumulatorResult("num-lines");
System.out.println("Total lines: " + lines);

注意：累加器结果在作业结束后才可获取（同步 execute()）。对于长期运行的流任务，想要实时可见的统计，请结合 Flink Metrics + Prometheus/Grafana，或周期性触发 savepoint/检查点对齐的离线读取方案。

命名空间合并

同一作业内，多个算子只要使用相同名称注册的累加器，Flink 会自动合并结果（方便全局统计）。

4、自定义累加器：实现可复用的聚合逻辑

两种接口可选：

• Accumulator<V, R>：添加值类型 V 与 最终结果类型 R 可不同（更灵活）
• SimpleAccumulator<T>：添加与结果类型相同（适合计数器等）

自定义直方图（示例）

public class IntHistogram implements Accumulator<Integer, Map<Integer, Integer>> {
  private final Map<Integer, Integer> bins = new HashMap<>();

  @Override
  public void add(Integer value) {
    bins.merge(value, 1, Integer::sum);
  }

  @Override
  public Map<Integer, Integer> getLocalValue() { return bins; }

  @Override
  public void resetLocal() { bins.clear(); }

  @Override
  public void merge(Accumulator<Integer, Map<Integer, Integer>> other) {
    other.getLocalValue().forEach((k, v) -> bins.merge(k, v, Integer::sum));
  }

  @Override
  public Accumulator<Integer, Map<Integer, Integer>> clone() {
    IntHistogram copy = new IntHistogram();
    copy.bins.putAll(this.bins);
    return copy;
  }
}

注册与使用：

public static class HistFunction extends RichFlatMapFunction<String, String> {
  private final IntHistogram hist = new IntHistogram();

  @Override
  public void open(Configuration parameters) {
    getRuntimeContext().addAccumulator("len-hist", hist);
  }

  @Override
  public void flatMap(String value, Collector<String> out) {
    hist.add(value.length());
    out.collect(value);
  }
}

5、实战模板与常见坑

5.1 模板：带累加器的 RichMapFunction

public static class ParsingMap extends RichMapFunction<String, Integer> {
  private final LongCounter parseOk = new LongCounter();
  private final LongCounter parseErr = new LongCounter();

  @Override
  public void open(Configuration parameters) {
    getRuntimeContext().addAccumulator("parse-ok", parseOk);
    getRuntimeContext().addAccumulator("parse-err", parseErr);
  }

  @Override
  public Integer map(String value) {
    try {
      int n = Integer.parseInt(value);
      parseOk.add(1);
      return n;
    } catch (Exception e) {
      parseErr.add(1);
      return 0;
    }
  }
}

5.2 常见坑与规避

1. 期望“实时”看到累加器值

   • 累加器只能在作业结束后获取。需要在线可见 → 用 Metrics。

2. Lambda 写复杂逻辑

   • 排查困难、不可复用。复杂逻辑建议用命名类或 Rich Function。

3. 忘记在 open() 中注册累加器

   • addAccumulator() 需要在 Rich Function 的 open() 调用，否则不会出现在结果里。

4. 命名冲突

   • 全作业是同一命名空间，不同用途请使用唯一名称，避免混淆合并。

5. 大对象在累加器中堆积

   • 累加器内容会随作业合并并回传，尽量存轻量聚合值（计数/分桶），避免大体量对象。

6. 流作业同步等待

   • env.execute() 同步等待长期流作业并不现实。若仅为调试统计，考虑临时跑一小段、或在测试环境使用 bounded 输入。

6、结语与建议

• UDF 选型：简单逻辑用 Lambda；需要生命周期、上下文、指标/累加器等工程化能力就用 Rich Function。
• 累加器定位：更适合离线统计与调试回放；在线可观测用 Metrics。
• 工程化落地：统一封装 Rich UDF（open/close 里做资源管理与累加器注册），提供明确的命名规范与监控方案。