大家有多少使用Flink处理批数据的？

Question

Flink虽然号称流批一体，但在处理离线批数据时，有些功能还是不支持的，如做分组统计，DataStreamApi如果做分组统计必须要有开窗函数,而批处理无需windows函数，只能用DataSet API实现，不知大家如何处理的？

Answer 1

引自免费微信小程序：皆我百晓生

Apache Flink 提供了 DataStream API 和 DataSet API 两种编程接口，确实如你所说，它们在处理流数据和批数据时具有不同的特点。DataStream API 面向实时流处理，而 DataSet API 则面向批量处理场景。在批处理场景下，虽然 Flink 主推的确实是流批一体的 DataStream API，但在早期版本中（例如 1.9 版本之前），DataSet API 更适合做无窗口的分组统计等操作。

对于分组统计的需求，在 Flink 的 DataStream API 中，确实需要结合 windowing 来实现类似功能，这是因为流处理的本质要求对无界数据流进行有界处理。不过，自 Flink 1.9 版本开始，DataStream API 引入了 BoundedOutOfOrderness 窗口以及 ProcessFunction 等机制，可以更加灵活地处理批数据或接近批的数据场景，甚至在某些情况下，可以通过设置非常大的时间窗口来模拟批处理的效果。

而对于纯粹的批处理场景，尽管 Flink 推荐使用统一的 DataStream API（在新版本中已经足够强大以处理批数据），但如果你依然希望使用 DataSet API，以下是一个使用 DataSet API 实现分组统计的示例：

import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;

public class BatchProcessingExample {

    public static void main(String[] args) throws Exception {
        // 创建执行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        // 假设有一个 CSV 数据源
        DataSet<String> text = env.readTextFile("/path/to/input.csv");

        // 将数据转换为 Tuple，并进行分组和求和统计
        DataSet<Tuple2<String, Integer>> counts = text
                .map(line -> { // 这里是将每行文本解析成 (key, value) 对
                    String[] words = line.split(",");
                    return new Tuple2<>(words[0], Integer.parseInt(words[1]));
                })
                .groupBy(0)
                .sum(1);

        // 打印结果
        counts.print();

        // 执行作业
        env.execute("Batch Processing Example");
    }
}

然而，鉴于 Flink 社区当前的发展趋势，建议尽量采用 DataStream API 进行批处理，因为这有助于简化代码结构并保持未来兼容性。在 DataStream API 中处理静态批数据集，你可以通过创建一个无边界的全局窗口来达到类似于批处理的效果：

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;

public class BatchProcessingWithDataStreamAPI {

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 假设数据源（这里为了演示，用 fromElements）
        DataStream<String> text = env.fromElements(
                "apple,10",
                "banana,20",
                "apple,30",
                "orange,40"
        );

        DataStream<Tuple2<String, Integer>> counts = text
                .map(new MapFunction<String, Tuple2<String, Integer>>() {
                    @Override
                    public Tuple2<String, Integer> map(String value) {
                        String[] words = value.split(",");
                        return new Tuple2<>(words[0], Integer.parseInt(words[1]));
                    }
                })
                .keyBy(0)
                .timeWindowAll(Time.unbounded()) // 使用无边界全局窗口
                .sum(1); 

        counts.print().setParallelism(1); // 设置并行度为 1，避免重复打印

        env.execute("Batch Processing with DataStream API");
    }
}

以上代码展示了如何利用 DataStream API 中的无边界全局窗口来近似模拟批处理的分组统计行为。当然，在实际应用中，如果不需要事件时间处理或者窗口特性，通常无需指定任何窗口，直接对 KeyedStream 进行聚合操作即可。对于不涉及时间概念的纯批处理任务，DataStream API 已经能够很好地胜任。