Flink DataStream API

Flink DataStream API 2019-07-15 02:07:08 320人浏览无得

摘要

1. api基本概念 flink程序可以对分布式集合进行转换（例如： filtering, mapping, updating state, joining, grouping, defining windows, aggreg

1. api基本概念

flink程序可以对分布式集合进行转换（例如： filtering, mapping, updating state, joining, grouping, defining windows, aggregating）

集合最初是从源创建的（例如，从文件、kafka主题或本地内存集合中读取）

结果通过sink返回，例如，可以将数据写入(分布式)文件，或者写入标准输出(例如，命令行终端)

根据数据源的类型（有界或无界数据源），可以编写批处理程序或流处理程序，其中使用DataSet API进行批处理，并使用DataStream API进行流处理。

Flink有特殊的类DataSet和DataStream来表示程序中的数据。在DataSet的情况下，数据是有限的，而对于DataStream，元素的数量可以是无限的。

Flink程序看起来像转换数据集合的常规程序。每个程序都包含相同的基本部分：

获取一个执行环境
加载/创建初始数据
指定数据上的转换
指定计算结果放在哪里
触发程序执行

为了方便演示，先创建一个项目，可以从Maven模板创建，例如：

mvn archetype:generate 
      -DarchetypeGroupId=org.apache.flink 
      -DarchetypeArtifactId=flink-quickstart-java 
      -DarchetypeVersion=1.10.0 
      -DgroupId=com.cjs.example 
      -dartifactId=flink-quickstart 
      -Dversion=1.0.0-SNAPSHOT 
      -Dpackage=com.cjs.example.flink 
      -DinteractiveMode=false

也可以直接创建SpringBoot项目，自行引入依赖：

<dependency>
    <groupId>org.apache.flinkgroupId>
    <artifactId>flink-javaartifactId>
    <version>1.10.0version>
    <scope>providedscope>
dependency>
<dependency>
    <groupId>org.apache.flinkgroupId>
    <artifactId>flink-streaming-java_2.11artifactId>
    <version>1.10.0version>
    <scope>providedscope>
dependency>
<dependency>
    <groupId>org.apache.flinkgroupId>
    <artifactId>flink-connector-kafka-0.10_2.11artifactId>
    <version>1.10.0version>
dependency>

StreamExecutionEnvironment是所有Flink程序的基础。你可以在StreamExecutionEnvironment上使用以下静态方法获得一个：

getExecutionEnvironment()

createLocalEnvironment()

createRemoteEnvironment(String host, int port, String... jarFiles)

通常，只需要使用getExecutionEnvironment()即可，因为该方法会根据上下文自动推断出当前的执行环境

从文件中读取数据，例如：

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

DataStream text = env.readTextFile("file:///path/to/file");

对DataStream应用转换，例如：

DataStream input = ...;

DataStream parsed = input.map(new MapFunction() {
    @Override
    public Integer map(String value) {
        return Integer.parseInt(value);
    }
});

通过创建一个sink将结果输出，例如：

writeAsText(String path)

print()

最后，调用StreamExecutionEnvironment上的execute()执行：

//  Triggers the program execution
env.execute();

//  Triggers the program execution asynchronously
final JobClient jobClient = env.executeAsync();
final JobExecutionResult jobExecutionResult = jobClient.getJobExecutionResult(userClassloader).get();

下面通过单词统计的例子来加深对这一流程的理解，WordCount程序之于大数据就相当于是HelloWorld之于Java，哈哈哈

package com.cjs.example.flink;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;


public class WordCount {
    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        DataSet text = env.readTextFile("/Users/asdf/Desktop/input.txt");
        DataSet> counts =
                // split up the lines in pairs (2-tuples) containing: (word,1)
                text.flatMap(new Tokenizer())
                        // group by the tuple field "0" and sum up tuple field "1"
                        .groupBy(0)
                        .sum(1);
        
        counts.writeAsCsv("/Users/asdf/Desktop/aaa", "
", " ");
        env.execute();
    }

    static class Tokenizer implements FlatMapFunction> {
        @Override
        public void flatMap(String value, Collector> out) throws Exception {
            // nORMalize and split the line
            String[] tokens = value.toLowerCase().split("\W+");

            // emit the pairs
            for (String token : tokens) {
                if (token.length() > 0) {
                    out.collect(new Tuple2<>(token, 1));
                }
            }
        }
    }
}

为Tuple定义keys

python中也有Tuple（元组）

DataStream> input = // [...]
KeyedStream,Tuple> keyed = input.keyBy(0)

元组按第一个字段(整数类型的字段)分组

还可以使用POJO的属性来定义keys，例如：

// some ordinary POJO (Plain old Java Object)
public class WC {
  public String word;
  public int count;
}
DataStream words = // [...]
DataStream wordCounts = words.keyBy("word").window();

先来了解一下KeyedStream

因此可以通过KeySelector方法来自定义

// some ordinary POJO
public class WC {public String word; public int count;}
DataStream words = // [...]
KeyedStream keyed = words
  .keyBy(new KeySelector() {
     public String geTKEy(WC wc) { return wc.word; }
   });

如何指定转换方法呢？

方式一：匿名内部类

data.map(new MapFunction () {
    public Integer map(String value) { return Integer.parseInt(value); }
});

方式二：Lamda

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

2. DataStream API

下面这个例子，每10秒钟统计一次来自WEB Socket的单词次数

package com.cjs.example.flink;

import org.apache.flink.api.common.functions.FlatMapFunction;
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;
import org.apache.flink.util.Collector;

public class WindowWordCount {

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

        DataStream> dataStream = env.socketTextStream("localhost", 9999)
                .flatMap(new Splitter())
                .keyBy(0)
                .timeWindow(Time.seconds(10))
                .sum(1);

        dataStream.print();

        env.execute("Window WordCount");
    }

    static class Splitter implements FlatMapFunction> {
        @Override
        public void flatMap(String value, Collector> out) throws Exception {
            String[] words = value.split("\W+");
            for (String word : words) {
                out.collect(new Tuple2(word, 1));
            }
        }
    }
}