Flink DataStream API 编程指南

实例程序

public class WindowWorkCount {

    public static void main(String[] args) throws Exception {
				// 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
				env.enableCheckpointing(5000);
				env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
				env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

				// 构建数据流
        DataStream<Tuple2<String, Integer>> dataStream = env
                .socketTextStream("127.0.0.1", 9999)  // 监听sockt文本输入流
                .flatMap(new Splitter())              // 合并输入映射
                .keyBy(value -> value.f0)             // 定义元素键
                .timeWindow(Time.seconds(5))          // 定义时间窗口
                .sum(1);                              // 对元素键分组进行sum统计
        dataStream.print();
        env.execute("window wordcount");
    }

    public static class Splitter extends RichFlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
            for(String word : value.split(" ")){
                out.collect(new Tuple2<>(word, 1));
            }
        }
    }
}

启动 nc 监听输入

nc -lk 9999

Data Sources 数据源

数据源源是程序读取输入的地方。可以使用 StreamExecutionEnvironment. addSource(sourceFunction) 将一个源附加到程序中。

Flink 提供了许多预先实现的数据源函数，同时也支持自定义实现数据源。

→ File-based 文件数据源

• readTextFile(path) ：逐行读取文件文本，该文件文本需要符合 TextInputFormat 规范，并将其作为字符串返回；
• readFile(fileInputFormat, path) ：根据文件输入格式一次性读取文本；
• readFile(fileInputFormat, path, watchType, interval, pathFilter, typeInfo) ：根据提供的 watchType，以 interval 指定间隔（ms），读取指定文件路径文件文本，通过 pathFilter 用户可以进一步从文件中排除文件。 其中typeInfo 参数包含以下：
  • FileProcessingMode.PROCESS_CONTINUOUSLY：持续监视文件新数据；
  • FileProcessingMode.PROCESS_ONCE：一次处理当前路径中的数据后退出；

→ Socket-based 套接字数据源

• socketTextStream：从套接字读取文本，元素可以指定分隔符；

→ Collection-based 集合数据源