Flink之DataStreamApi编程实战

古城微笑少年丶 2023-10-06 18:09 20阅读 0赞

#### 一、Flink DataStreamApi作用和使用场景 ####

Flink中的DataStream程序是实现在数据流上的transformation(如filtering，updating state， defining windows，aggregating)的普通程序。创建数据流的来源多种多样（如消息队列，socket流，文件等）。程序通过data sink返回结果，如将数据写入文件，或发送到标准输出（如命令行终端）。Flink程序可以在多种上下文中运行，如独立运行或是嵌入在其他程序中执行。程序的执行可以发生在本地JVM，或者在一个拥有许多设备的集群上。

##### 二、 Flink DataStreamApi 常用函数 #####

###### 2.1 Map ######

DataStream -> DataStream

获取一个element并产出一个element。下例是一个将输入*2的map方法：
     
    
    DataStream<Integer> dataStream = //...
    dataStream.map(new MapFunction<Integer, Integer>() {
        
    　　@Override
    　　public Integer map(Integer value) throws Exception {
        
    　　　　return 2 * value;
    　　}
    });

###### 2.2 FlapMap ######

DataStream -> DataStream

获取一个element，并产生出0、1或多个element。下例是一个为句子分词的flatmap方法
    
     
    
    dataStream.flatMap(new FlatMapFunction<String, String>() {
        
    　　@Override
    　　public void flatMap(String value, Collector<String> out) throws Exception {
        
    　　　　for(String word: value.split(" ")){
        
    　　　　out.collect(word);
    　　　　}
    　　}
    });

###### 2.3 Filter ######

在每个获取的element上运行一个boolean方法，留下那些方法返回true的element。下例是一个过滤掉0值的filter
     
    
    dataStream.filter(new FilterFunction<Integer>() {
        
    　　@Override
    　　public boolean filter(Integer value) throws Exception {
        
    　　　　return value != 0;
    　　}
    });

###### 2.4 KeyBy ######

DataStream -> KeyedStream

将流逻辑分为不相交的分区，每个分区包含的都是具有相同key的element，该分区方法使用hash分区实现。定义key的方法见于Keys。下例是一个返回KeyedDataStream的transformation。
     
    
    dataStream.keyBy("someKey") // Key by field "someKey"
    dataStream.keyBy(0) // Key by the first element of a Tuple

###### 2.5 Reduce ######

KeyedStream -> DataStream

一个在keyed data stream上“滚动”进行的reduce方法。将上一个reduce过的值和当前element结合，产生新的值并发送出。下例是一个创建部分和的reduce方法。
     
    
    keyedStream.reduce(new ReduceFunction<Integer>() {
        
    　　@Override
    　　public Integer reduce(Integer value1, Integer value2) throws Exception {
        
    　　　　return value1 + value2;
    　　}
    });

###### 2.6 Fold ######

KeyedStream -> DataStream

一个在带有初始值的数据流上“滚动”进行的fold方法。将上一个fold的值和当前element结合，产生新的值并发送出。下例是一个fold方法，当应用于序列{
        1, 2, 3, 4, 5}时，它发出序列{
        "start-1", "start-1-2", "start-1-2-3" …}。
     
    
    DataStream<String> result = keyedStream.fold("start", new FoldFunction<Integer, String>() {
        
    　　@Override
    　　public String fold(String current, Integer value) {
        
    　　　　return current + "-" + value;
    　　}
    });

###### 2.7 Aggregations ######

KeyedStream -> DataStream

在一个keyed DataStream上“滚动”进行聚合的方法。其中，min和minBy的区别在于min返回最小值，而minBy返回的是带有在此域中最小值的element（max和maxBy一样如此）。
     
    
    keyedStream.sum(0);
    keyedStream.sum("key");
    keyedStream.min(0);
    keyedStream.min("key");
    keyedStream.max(0);
    keyedStream.max("key");
    keyedStream.minBy(0);
    keyedStream.minBy("key");
    keyedStream.maxBy(0);
    keyedStream.maxBy("key");

###### 2.8 Window ######

KeyedStream - > WindowedStream

Window可以定义在已经分区的KeyedStream上。窗口将根据一些特征（如最近5秒到达的数据）将数据按其各自的key集合在一起。有关窗口的完整描述见于windows
     
    
    // Last 5 seconds of data
    
    dataStream.keyBy(0).window(TumblingEventTimeWindows.of(Time.seconds(5)));

###### 2.9 WindowAll ######

DataStream -> AllWindowedStream

Window可以定义在普通的DataStream上。窗口将根据一些特征（如最近5秒到达的数据）将所有Stream事件集合在一起。有关窗口的完整描述见于windows
    警告：该transformation在很多情况下都不是并行化的，所有数据将被收集到一个运行windowAll Operator的任务上。
    
     
    
    dataStream.windowAll(TumblingEventTimeWindows.of(Time.seconds(5))); // Last 5 seconds of data

###### 2.10 Window Apply ######

WindowedStream -> DataStream

AllWindowedStream -> DataStream

将一个一般函数应用到window整体上去，下面是一个人工计算window中所有element的总和的应用。
    注意：如果你正在使用一个windowAll的transformation，你需要使用AllWindowFunction来代替下例中的参数。
     
    
    windowedStream.apply (new WindowFunction<Tuple2<String,Integer>, Integer, Tuple, Window>() {
        
    public void apply (Tuple tuple,
    　　Window window,
    　　Iterable<Tuple2<String, Integer>> values,
    　　Collector<Integer> out) throws Exception {
        
    　　　　int sum = 0;
    　　　　for (value t: values) {
        
    　　　　　　sum += t.f1;
    　　　　}
    　　　　out.collect (new Integer(sum));
    　　}
    });
    
    // applying an AllWindowFunction on non-keyed window stream
    allWindowedStream.apply (new AllWindowFunction<Tuple2<String,Integer>, Integer, Window>() {
        
    public void apply (Window window,
    　　Iterable<Tuple2<String, Integer>> values,
    　　Collector<Integer> out) throws Exception {
        
    　　　　int sum = 0;
    　　　　for (value t: values) {
        
    　　　　　　sum += t.f1;
    　　　　}
    　　　　out.collect (new Integer(sum));
    　　}
    });

###### 2.11 Window Reduce ######

WindowedStream -> DataStream

对窗口应用一个功能性reduce方法并返回reduce的结果
     
    
    windowedStream.reduce (new ReduceFunction<Tuple2<String,Integer>() {
        
    　　public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws Exception {
        
    　　return new Tuple2<String,Integer>(value1.f0, value1.f1 + value2.f1);
    　　}
    };

###### 2.12 Window Fold ######

Windowed Stream -> DataStream

对窗口应用一个功能性fold方法。下例代码在应用到序列(1, 2, 3, 4, 5)时，它将该序列fold成为字符串"start-1-2-3-4-5"
     
    
    windowedStream.fold("start-", new FoldFunction<Integer, String>() {
        
    　　public String fold(String current, Integer value) {
        
    　　　　return current + "-" + value;
    　　}
    };