Flink流处理API
B / Flink

Flink流处理API

5.1、Environment

5.1.1 getExecutionEnvironment

创建一个执行环境，表示当前执行程序的上下文。 如果程序是独立调用的，则 此方法返回本地执行环境；如果从命令行客户端调用程序以提交到集群，则此方法 返回此集群的执行环境，也就是说，getExecutionEnvironment 会根据查询运行的方 式决定返回什么样的运行环境，是最常用的一种创建执行环境的方式。

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); 

StreamExecutionEnvironment env = 
StreamExecutionEnvironment.getExecutionEnvironment();

如果没有设置并行度，会以 flink-conf.yaml 中的配置为准，默认是 1。

5.1.2、createLocalEnvironment

返回本地执行环境，需要在调用时指定默认的并行度。

LocalStreamEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(1);

5.1.3、createRemoteEnvironment

返回集群执行环境，将 Jar 提交到远程服务器。需要在调用时指定 JobManager 的 IP 和端口号，并指定要在集群中运行的 Jar 包。

StreamExecutionEnvironment env = 
StreamExecutionEnvironment.createRemoteEnvironment("jobmanage-hostname", 6123, 
"YOURPATH//WordCount.jar");

5.2、Source

5.2.1 从集合读取数据

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

        // 从集合中读取数据
        DataStream<SensorReading> dataStream = env.fromCollection(Arrays.asList(
                new SensorReading("sensor_1", 1547718199L, 35.8),
                new SensorReading("sensor_6", 1547718201L, 15.4),
                new SensorReading("sensor_7", 1547718202L, 6.7),
                new SensorReading("sensor_10", 1547718205L, 38.1)
        ));

        DataStream<Integer> integerDataStream = env.fromElements(1, 2, 4, 67, 189);

        // 打印输出
        dataStream.print("data");
        integerDataStream.print("int");

        // 执行
        env.execute();
    }
}

5.2.2、从文件读取数据

DataStream<String> dataStream = env.readTextFile("YOUR_FILE_PATH ");

5.2.3、以 kafka 消息队列的数据作为来源

需要引入 kafka 连接器的依赖：

<! -- https://mvnrepository.com/artifa ct/org.apache.flink/flink - connector - kafka - 0.11 -- > 
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka-0.11_2.12</artifactId>
    <version>1.10.1</version>
</dependency>

具体代码如下：

// kafka
配置项 
Properties properties = new Properties(); 
properties.setProperty("bootstrap.servers", "localhost:9092"); 
properties.setProperty("group.id", "consumer-group"); 
properties.setProperty("key.deserializer", 
"org.apache.kafka.common.serialization.StringDeserializer"); 
properties.setProperty("value.deserializer", 
"org.apache.kafka.common.serialization.StringDeserializer"); 
properties.setProperty("auto.offset.reset", "latest"); 
 
 
// 从 kafka读取数据 
DataStream<String> dataStream = env.addSource( new 
FlinkKafkaConsumer011<String>("sensor", new SimpleStringSchema(), properties));

5.2.4、自定义 Source

除了以上的 source 数据来源，我们还可以自定义 source。需要做的，只是传入一个 SourceFunction 就可以。具体调用如下：

DataStream<SensorReading> dataStream = env.addSource( new MySensor());

我们希望可以随机生成传感器数据，MySensorSource 具体的代码实现如下：

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

        // 从文件读取数据
        DataStream<SensorReading> dataStream = env.addSource( new MySensorSource() );

        // 打印输出
        dataStream.print();

        env.execute();
    }

    // 实现自定义的SourceFunction
    public static class MySensorSource implements SourceFunction<SensorReading>{
        // 定义一个标识位，用来控制数据的产生
        private boolean running = true;

        @Override
        public void run(SourceContext<SensorReading> ctx) throws Exception {
            // 定义一个随机数发生器
            Random random = new Random();

            // 设置10个传感器的初始温度
            HashMap<String, Double> sensorTempMap = new HashMap<>();
            for( int i = 0; i < 10; i++ ){
                sensorTempMap.put("sensor_" + (i+1), 60 + random.nextGaussian() * 20);
            }

            while (running){
                for( String sensorId: sensorTempMap.keySet() ){
                    // 在当前温度基础上随机波动
                    Double newtemp = sensorTempMap.get(sensorId) + random.nextGaussian();
                    sensorTempMap.put(sensorId, newtemp);
                    ctx.collect(new SensorReading(sensorId, System.currentTimeMillis(), newtemp));
                }
                // 控制输出频率
                Thread.sleep(1000L);
            }
        }

        @Override
        public void cancel() {
            running = false;
        }
    }
}

5.3、Transform

转换算子

5.3.1、map

DataStream<Integer> mapStram = dataStream.map(new MapFunction<String, Integer>() { 
    public Integer map(String value) throws Exception { 
        return value.length(); 
    } 
});

5.3.2、flatMap

DataStream<String> flatMapStream = dataStream.flatMap(new FlatMapFunction<String, 
String>() { 
    public void flatMap(String value, Collector<String> out) throws Exception { 
        String[] fields = value.split(",");  
        for( String field: fields ) 
            out.collect(field); 
 } 
});

5.3.3、Filter

DataStream<Interger> filterStream = dataStream.filter(new FilterFunction<String>() 
{ 
    public boolean filter(String value) throws Exception { 
        return value == 1; 
    } 
});

5.3.4、KeyBy


DataStream → KeyedStream：逻辑地将一个流拆分成不相交的分区，每个分区包含具有相同 key 的元素，在内部以 hash 的形式实现的。

5.3.5、滚动聚合算子（Rolling Aggregation）

这些算子可以针对 KeyedStream 的每一个支流做聚合。

sum() 
min() 
max() 
minBy() 
maxBy()

5.3.6、Reduce

KeyedStream → DataStream：一个分组数据流的聚合操作，合并当前的元素和上次聚合的结果，产生一个新的值，返回的流中包含每一次聚合的结果，而不是只返回最后一次聚合的最终结果。

DataStream<String> inputStream = env.readTextFile("sensor.txt"); 
 
        // 转换成 SensorReading类型 
        DataStream<SensorReading> dataStream = inputStream.map(new MapFunction<String, SensorReading>() { 
            public SensorReading map(String value) throws Exception { 
                String[] fileds = value.split(","); 
                return new SensorReading(fileds[0], new Long(fileds[1]), new Double(fileds[2])); 
            } 
        }); 
 
        // 分组         
        KeyedStream<SensorReading, Tuple> keyedStream = dataStream.keyBy("id"); 
 
        // reduce聚合，取最小的温度值，并输出当前的时间戳 
        DataStream<SensorReading> reduceStream = keyedStream.reduce(new ReduceFunction<SensorReading>() { 
            @Override 
            public SensorReading reduce(SensorReading value1, SensorReading value2) throws Exception { 
                return new SensorReading( 
                        value1.getId(), 
                        value2.getTimestamp(), 
                        Math.min(value1.getTemperature(), value2.getTemperature())); 
            } 
        });

5.3.7、Split 和 Select


DataStream → SplitStream：根据某些特征把一个 DataStream 拆分成两个或者多个 DataStream。


SplitStream→DataStream：从一个 SplitStream 中获取一个或者多个DataStream。
需求：传感器数据按照温度高低（以 30 度为界），拆分成两个流。

SplitStream<SensorReading> splitStream = dataStream.split(new 
OutputSelector<SensorReading>() { 
    @Override 
    public Iterable<String> select(SensorReading value) { 
        return (value.getTemperature() > 30) ? Collections.singletonList("high") : 
Collections.singletonList("low"); 
    } 
}); 
 
DataStream<SensorReading> highTempStream = splitStream.select("high"); 
DataStream<SensorReading> lowTempStream = splitStream.select("low"); 
DataStream<SensorReading> allTempStream = splitStream.select("high", "low");

5.3.8、Connect 和 CoMap

DataStream,DataStream → ConnectedStreams：连接两个保持他们类型的数据流，两个数据流被 Connect 之后，只是被放在了一个同一个流中，内部依然保持各自的数据和形式不发生任何变化，两个流相互独立。

CoMap,CoFlatMap

ConnectedStreams → DataStream：作用于 ConnectedStreams 上，功能与 map和 flatMap 一样，对 ConnectedStreams 中的每一个 Stream 分别进行 map 和 flatMap处理。

// 合流 connect 
DataStream<Tuple2<String, Double>> warningStream = highTempStream.map(new 
MapFunction<SensorReading, Tuple2<String, Double>>() { 
    @Override 
     public Tuple2<String, Double> map(SensorReading value) throws Exception { 
            return new Tuple2<>(value.getId(), value.getTemperature()); 
        } 
    }); 
    ConnectedStreams<Tuple2<String, Double>, SensorReading> connectedStreams = 
    warningStream.connect(lowTempStream); 
     
    DataStream<Object> resultStream = connectedStreams.map(new 
    CoMapFunction<Tuple2<String,Double>, SensorReading, Object>() { 
        @Override 
        public Object map1(Tuple2<String, Double> value) throws Exception { 
            return new Tuple3<>(value.f0, value.f1, "warning"); 
        } 
     
        @Override 
        public Object map2(SensorReading value) throws Exception { 
            return new Tuple2<>(value.getId(), "healthy"); 
        } 
});

5.3.9、Union

DataStream → DataStream：对两个或者两个以上的 DataStream 进行 union 操作，产生一个包含所有 DataStream 元素的新 DataStream。

DataStream<SensorReading> unionStream = highTempStream.union(lowTempStream);

Connect 与 Union 区别：
1、Union 之前两个流的类型必须是一样，Connect 可以不一样，在之后的 coMap中再去调整成为一样的。
2、Connect 只能操作两个流，Union 可以操作多个。

5.4、支持的数据类型

Flink 流应用程序处理的是以数据对象表示的事件流。所以在 Flink 内部，我们需要能够处理这些对象。它们需要被序列化和反序列化，以便通过网络传送它们；或者从状态后端、检查点和保存点读取它们。为了有效地做到这一点，Flink 需要明确知道应用程序所处理的数据类型。Flink 使用类型信息的概念来表示数据类型，并为每个数据类型生成特定的序列化器、反序列化器和比较器。

Flink 还具有一个类型提取系统，该系统分析函数的输入和返回类型，以自动获取类型信息，从而获得序列化器和反序列化器。但是，在某些情况下，例如 lambda函数或泛型类型，需要显式地提供类型信息，才能使应用程序正常工作或提高其性能。

Flink 支持 Java 和 Scala 中所有常见数据类型。使用最广泛的类型有以下几种。

5.4.1、基础数据类型

Flink 支持所有的 Java 和 Scala 基础数据类型，Int, Double, Long, String, …

DataStream<Integer> numberStream = env.fromElements(1, 2, 3, 4); 
numberStream.map(data -> data * 2);

5.4.2、Java 和 Scala 元组（Tuples）

DataStream<Tuple2<String, Integer>> personStream = env.fromElements( 
        new Tuple2("Adam", 17), 
        new Tuple2("Sarah", 23) ); 
personStream.filter(p -> p.f1 > 18);

5.4.3、Scala 样例类（case classes）

case class Person(name: String, age: Int)  
val persons: DataStream[Person] = env.fromElements( 
    Person("Adam", 17),  
    Person("Sarah", 23)) 
persons.filter(p => p.age > 18)

5.4.4、Java 简单对象（POJOs）

public class Person { 
public String name; 
public int age; 
  public Person() {} 
  public Person(String name, int age) {  
    this.name = name;       
    this.age = age;   
  } 
} 
DataStream<Person> persons = env.fromElements(    
new Person("Alex", 42),    
new Person("Wendy", 23));

5.4.5、其它（Arrays, Lists, Maps, Enums, 等等）

Flink 对 Java 和 Scala 中的一些特殊目的的类型也都是支持的，比如 Java 的ArrayList，HashMap，Enum 等等。

5.5、实现UDF 函数——更细粒度的控制流

5.5.1、函数类（Function Classes）

Flink 暴露了所有 udf 函数的接口(实现方式为接口或者抽象类)。例如MapFunction, FilterFunction, ProcessFunction 等等。

下面例子实现了 FilterFunction 接口：

DataStream<String> flinkTweets = tweets.filter(new FlinkFilter()); 
 
public static class FlinkFilter implements FilterFunction<String> { 
    @Override 
    public boolean filter(String value) throws Exception { 
        return value.contains("flink"); 
 } 
}

还可以将函数实现成匿名类

DataStream<String> flinkTweets = tweets.filter(new FilterFunction<String>() { 
    @Override 
    public boolean filter(String value) throws Exception { 
        return value.contains("flink"); 
    } 
});

我们 filter 的字符串"flink"还可以当作参数传进去。

DataStream<String> tweets = env.readTextFile("INPUT_FILE "); 
 
DataStream<String> flinkTweets = tweets.filter(new KeyWordFilter("flink")); 
 
public static class KeyWordFilter implements FilterFunction<String> { 
    private String keyWord; 
 
    KeyWordFilter(String keyWord) { this.keyWord = keyWord; } 
 
    @Override 
    public boolean filter(String value) throws Exception { 
        return value.contains(this.keyWord); 
    } 
}

5.5.2、匿名函数（Lambda Functions）

DataStream<String> tweets = env.readTextFile("INPUT_FILE"); 
 
DataStream<String> flinkTweets = tweets.filter( tweet -> tweet.contains("flink") );

5.5.3、富函数（Rich Functions）

“富函数”是 DataStream API 提供的一个函数类的接口，所有 Flink 函数类都有其 Rich 版本。它与常规函数的不同在于，可以获取运行环境的上下文，并拥有一些生命周期方法，所以可以实现更复杂的功能。

  • RichMapFunction
  • RichFlatMapFunction
  • RichFilterFunction

Rich Function 有一个生命周期的概念。典型的生命周期方法有：

  • open()方法是 rich function 的初始化方法，当一个算子例如 map 或者 filter被调用之前 open()会被调用。
  • close()方法是生命周期中的最后一个调用的方法，做一些清理工作。
  • getRuntimeContext()方法提供了函数的 RuntimeContext 的一些信息，例如函数执行的并行度，任务的名字，以及 state 状态
public static class MyMapFunction extends RichMapFunction<SensorReading, 
Tuple2<Integer, String>> { 
    @Override 
    public Tuple2<Integer, String> map(SensorReading value) throws Exception { 
        return new Tuple2<>(getRuntimeContext().getIndexOfThisSubtask(), 
value.getId()); 
    } 
 
    @Override 
    public void open(Configuration parameters) throws Exception { 
        System.out.println("my map open"); 
        // 以下可以做一些初始化工作，例如建立一个和 HDFS的连接 
    } 
 
    @Override 
    public void close() throws Exception { 
        System.out.println("my map close"); 
        // 以下做一些清理工作，例如断开和 HDFS的连接 
    } 
}

5.6、Sink

Flink 没有类似于 spark 中 foreach 方法，让用户进行迭代的操作。虽有对外的输出操作都要利用 Sink 完成。最后通过类似如下方式完成整个任务最终输出操作。

stream.addSink(new MySink(xxxx))

官方提供了一部分的框架的 sink。除此以外，需要用户自定义实现 sink。

5.6.1、Kafka

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka-0.11_2.12</artifactId>
    <version>1.10.1</version>
</dependency>

主函数中添加 sink：

dataStream.addSink(new FlinkKafkaProducer011[String]("localhost:9092", 
"test", new SimpleStringSchema()))

5.6.2、Redis

<dependency>
    <groupId>org.apache.bahir</groupId>
    <artifactId>flink-connector-redis_2.11</artifactId>
    <version>1.0</version>
</dependency>

定义一个 redis 的 mapper 类，用于定义保存到 redis 时调用的命令：

public static class MyRedisMapper implements RedisMapper<SensorReading>{ 
 
    // 保存到 redis的命令，存成哈希表  
    public RedisCommandDescription getCommandDescription() { 
        return new RedisCommandDescription(RedisCommand.HSET, "sensor_tempe"); 
    } 
 
    public String getKeyFromData(SensorReading data) { 
        return data.getId(); 
    } 
 
    public String getValueFromData(SensorReading data) { 
        return data.getTemperature().toString(); 
    } 
}

在主函数中调用：

FlinkJedisPoolConfig config = new FlinkJedisPoolConfig.Builder() 
        .setHost("localhost") 
        .setPort(6379) 
        .build(); 
dataStream.addSink( new RedisSink<SensorReading>(config, new MyRedisMapper()) );

5.6.3、Elasticsearch

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-elasticsearch6_2.12</artifactId>
    <version>1.10.1</version>
</dependency>

在主函数中调用：

// es的 httpHosts配置 
ArrayList<HttpHost> httpHosts = new ArrayList<>(); 
httpHosts.add(new HttpHost("localhost", 9200)); 
 
dataStream.addSink( new ElasticsearchSink.Builder<SensorReading>(httpHosts, new 
MyEsSinkFunction()).build());

ElasitcsearchSinkFunction 的实现：

public static class MyEsSinkFunction implements 
ElasticsearchSinkFunction<SensorReading>{ 
    @Override 
    public void process(SensorReading element, RuntimeContext ctx, RequestIndexer indexer) { 
 
        HashMap<String, String> dataSource = new HashMap<>(); 
        dataSource.put("id", element.getId()); 
        dataSource.put("ts", element.getTimestamp().toString()); 
        dataSource.put("temp", element.getTemperature().toString()); 
 
        IndexRequest indexRequest = Requests.indexRequest() 
                .index("sensor") 
                .type("readingData") 
                .source(dataSource); 
 
        indexer.add(indexRequest); 
    } 
}

5.6.4、JDBC 自定义 sink

<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>5.1.44</version>
</dependency>

添加 MyJdbcSink

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

        // 从文件读取数据
//        DataStream<String> inputStream = env.readTextFile("D:\\Projects\\BigData\\FlinkTutorial\\src\\main\\resources\\sensor.txt");
//
//        // 转换成SensorReading类型
//        DataStream<SensorReading> dataStream = inputStream.map(line -> {
//            String[] fields = line.split(",");
//            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
//        });

        DataStream<SensorReading> dataStream = env.addSource(new SourceTest4_UDF.MySensorSource());

        dataStream.addSink(new MyJdbcSink());

        env.execute();
    }

    // 实现自定义的SinkFunction
    public static class MyJdbcSink extends RichSinkFunction<SensorReading> {
        // 声明连接和预编译语句
        Connection connection = null;
        PreparedStatement insertStmt = null;
        PreparedStatement updateStmt = null;

        @Override
        public void open(Configuration parameters) throws Exception {
            connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/test", "root", "123456");
            insertStmt = connection.prepareStatement("insert into sensor_temp (id, temp) values (?, ?)");
            updateStmt = connection.prepareStatement("update sensor_temp set temp = ? where id = ?");
        }

        // 每来一条数据，调用连接，执行sql
        @Override
        public void invoke(SensorReading value, Context context) throws Exception {
            // 直接执行更新语句，如果没有更新那么就插入
            updateStmt.setDouble(1, value.getTemperature());
            updateStmt.setString(2, value.getId());
            updateStmt.execute();
            if( updateStmt.getUpdateCount() == 0 ){
                insertStmt.setString(1, value.getId());
                insertStmt.setDouble(2, value.getTemperature());
                insertStmt.execute();
            }
        }

        @Override
        public void close() throws Exception {
            insertStmt.close();
            updateStmt.close();
            connection.close();
        }
    }
}

评论

暂无

添加新评论