flink 学习（一）java 整合 flink__lrs_flink java

网络投稿 02-07 6211

目录前言一、流、批处理二、hello world1.环境2.流处理3.批处理4.自动处理三、文件数据源1.从文件中读取2.从目录中读取

前言

????????Apache Flink 是一个框架和分布式处理引擎，用于在无边界和有边界数据流上进行有状态的计算。Flink 能在所有常见集群环境中运行，并能以内存速度和任意规模进行计算。

一、流、批处理

????????有界流：有定义流的开始，但没有定义流的结束。它们会无休止地产生数据。无界流的数据必须持续处理，即数据被摄取后需要立刻处理。我们不能等到所有数据都到达再处理，因为输入是无限的，在任何时候输入都不会完成。

????????无界流：有定义流的开始，也有定义流的结束。有界流可以在摄取所有数据后再进行计算。有界流所有数据可以被排序，所以并不需要有序摄取。

????????流处理：无界流通常被称为流处理。当一条数据被处理完成后，序列化到缓存中，然后立刻通过网络传输到下一个节点，由下一个节点继续处理。数据是一条一条的处理，实时性高。

????????批处理：有界流处理通常被称为批处理。当一条数据被处理完成后，序列化到缓存中，并不会立刻通过网络传输到下一个节点，当缓存写满，就持久化到本地硬盘上，当所有数据都被处理完成后，才开始将处理后的一批数据通过网络传输到下一个节点。数据是一批一批处理，实时性低。

二、hello world 1.环境

jdk11 + flink1.14.4

引入flink 依赖

<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-java</artifactId> <version>1.14.4</version> </dependency>  <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java_2.12</artifactId> <version>1.14.4</version> </dependency>  <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-clients_2.12</artifactId> <version>1.14.4</version> </dependency> 2.流处理 @Test public void streamingTest() throws Exception { // flink 流执行环境 final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); //设置模式 STREAMING env.setRuntimeMode(RuntimeExecutionMode.STREAMING); //数据源，fromCollection env.fromCollection(List.of("nacos,python,java", "nacos,scripts,php", "nacos,java,springmvc", "nacos,sentinel,gateway")) //扁平化 .flatMap(new FlatMapFunction<String, String>() { @Override public void flatMap(String value, Collector<String> out) throws Exception { Arrays.stream(value.split(",")).forEach(v -> out.collect(v)); } }) //映射 .map(new MapFunction<String, Tuple2<String, Integer>>() { @Override public Tuple2<String, Integer> map(String value) throws Exception { return Tuple2.of(value, 1); } }) //分组 .keyBy((KeySelector<Tuple2<String, Integer>, String>) value -> value.f0) //求和 .sum(1) //打印结果 .print(); //开始执行 env.execute("flink streaming hello word"); }

执行结果：

6> (springmvc,1) 3> (python,1) 2> (java,1) 2> (java,2) 1> (scripts,1) 4> (php,1) 8> (nacos,1) 8> (nacos,2) 8> (nacos,3) 8> (nacos,4) 8> (sentinel,1) 8> (gateway,1)

可以看出，数据进行的是流处理：一条一条处理，一条一条打印

3.批处理

将执行模式修改为 BATCH

@Test public void batchTest() throws Exception { // flink 流执行环境 final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); //设置模式 BATCH env.setRuntimeMode(RuntimeExecutionMode.BATCH); ...

结果：

2> (java,2) 6> (springmvc,1) 3> (python,1) 8> (nacos,4) 1> (scripts,1) 4> (php,1) 8> (gateway,1) 8> (sentinel,1)

可以看出，数据进行的是批处理：数据处理完后，打印了统计结果，没有中间数据打印

4.自动处理

将执行模式修改为 AUTOMATIC

@Test public void automicTest() throws Exception { // flink 流执行环境 final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); //设置模式 AUTOMATIC env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC); ...

结果跟批处理一样，因为数据源是从集合中获取数据，数据是有界的，自动按照批处理模式进行处理。

三、文件数据源 1.从文件中读取

file.txt 文件中的内容：

java,python,c++ java,python,c# java,c++,php @Test public void fileTest() throws Exception { // flink 流执行环境 final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); //设置模式 AUTOMATIC env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC); //数据源，文件 env.readTextFile("E:\\tmp\\flink\\file.txt") //扁平化 .flatMap(new FlatMapFunction<String, String>() { @Override public void flatMap(String value, Collector<String> out) throws Exception { Arrays.stream(value.split(",")).forEach(v -> out.collect(v)); } }) //映射 .map(new MapFunction<String, Tuple2<String, Integer>>() { @Override public Tuple2<String, Integer> map(String value) throws Exception { return Tuple2.of(value, 1); } }) //分组 .keyBy((KeySelector<Tuple2<String, Integer>, String>) value -> value.f0) //求和 .sum(1) //打印结果 .print(); //开始执行 env.setParallelism(1); env.execute("flink streaming hello word1"); }

结果：一行一行读取

5> (c#,1) 2> (java,3) 4> (php,1) 3> (c++,2) 3> (python,2) 2.从目录中读取

目录中有两个文件，

文件1内容：

java,python,c++ java,python,c# java,c++,php

文件2内容：

java,python java,python java,python @Test public void dirTest() throws Exception { // flink 流执行环境 final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); //设置模式 AUTOMATIC env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC); //数据源，fromCollection env.readTextFile("E:\\tmp\\flink") //扁平化 .flatMap(new FlatMapFunction<String, String>() { @Override public void flatMap(String value, Collector<String> out) throws Exception { Arrays.stream(value.split(",")).forEach(v -> out.collect(v)); } }) //映射 .map(new MapFunction<String, Tuple2<String, Integer>>() { @Override public Tuple2<String, Integer> map(String value) throws Exception { return Tuple2.of(value, 1); } }) //分组 .keyBy((KeySelector<Tuple2<String, Integer>, String>) value -> value.f0) //求和 .sum(1) //打印结果 .print(); //开始执行 env.setParallelism(1); env.execute("flink streaming hello word1"); }

结果：

3> (c++,2) 3> (python,5) 5> (c#,1) 4> (php,1) 2> (java,6)