Flink入门 基础操作

这里写目录标题

• 一 、Flink入门
• • 1.1 Flink简介
  • • 1.1.1 处理无界和有界数据
    • 1.1.2 部署应用到任意地方
    • 1.1.3 运行任意规模应用
    • 1.1.4 利用内存性能
  • 1.2 Flink架构图
  • 1.3 入门案例演示
  • • 1.3.1 pom文件
    • 1.3.2 单词计数案例演示
    • 1.3.3 使用面向对象
    • 1.3.4 使用最佳实践
    • 1.3.5 抽离业务功能
  • 1.4 Flink核心概念
  • • 1.4.1 Flink概念
    • 1.4.2 本地观察Flink任务
    • 1.4.3 并行度
    • 1.4.4 数据传输策略
    • 1.4.5 Operator Chain
  • 1.5 Flink分布式运行环境
  • • 1.5.1 flink分布式四层模型
    • 1.5.2 Flink任务分布式运行流程
• 二、DataStream API
• • • 2.1 Flink之数据源
    • • 2.1.1 source简介
    • 2.1.2 数据源之collection
    • 2.1.3 自定义单并行度数据源
    • 2.1.4 自定义多并行度数据源
  • 2.2 常见Transformation操作
  • • 2.2.1 map和filter
    • 2.2.2 flatMap，keyBy和sum
    • 2.2.3 union
    • 2.2.4 connect, conMap和conFlatMap
    • 2.2.5 Split和Select
  • 2.3 常见sink操作
  • • 2.3.1 print() / printToErr()
    • 2.3.2 writeAsText()
    • 2.3.3 Flink提供的sink
• 三、DataSet API
• • 3.1 source
  • 3.2 transform
  • • 3.2.1. 算子概览
    • 3.2.3. MapPartition
    • 3.2.4. distinct
    • 3.2.5. join
    • 3.2.6. OutJoin
    • 3.2.7. Cross
    • 3.2.8. First-n 和 SortPartition
    • 3.2.9. partition
  • 3.3 sink
  • 3.4 Flink之广播变量
  • 3.5 Flink之Counter（计数器）

一 、Flink入门

1. 下载Flink版本 https://mirror.bit.edu.cn/apache/flink/flink-1.10.2/flink-1.10.2-bin-scala_2.11.tgz

1.1 Flink简介

Apache Flink® — Stateful Computations over Data Streams

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

---

Spark 很多技术参考了 Flink, 那为什么Spark更流行？

• Spark有一个好的产品经理
• Spark官方网站、Spark的任务界面、Spark Shell

Spark：顶层设计，认为 所有的数据都是批处理。
    如果数据量很大，处理速度很慢 ---> 离线处理任务
    数据量比较小，处理速度很快  ---> 微批处理 ，近实时处理任务 Sparking streaming
Flink：顶层设计，认为 所有的数据都是流。
    流是源源不断的，无界 ---> 实时处理任务
    10分钟，1小时，有界  ---> 离线处理任务

1.1.1 处理无界和有界数据

任何类型的数据都可以形成一种事件流。信用卡交易、传感器测量、机器日志、网站或移动应用程序上的用户交互记录，所有这些数据都形成一种流。

数据可以被作为 无界 或者 有界 流来处理。

1. 无界流(实时流，实时的程序)
   • 有定义流的开始，但没有定义流的结束。它们会无休止地产生数据。
   • 无界流的数据必须持续处理，即数据被摄取后需要立刻处理。我们不能等到所有数据都到达再处理，因为输入是无限的，在任何时候输入都不会完成。处理无界数据通常要求以特定顺序摄取事件，例如事件发生的顺序，以便能够推断结果的完整性。
2. 有界流
   • 有定义流的开始，也有定义流的结束。
   • 有界流可以在摄取所有数据后再进行计算。有界流所有数据可以被排序，所以并不需要有序摄取。有界流处理通常被称为批处理。

Apache Flink 擅长处理无界和有界数据集 精确的时间控制和状态化使得Flink的运行时(runtime)能够运行任何处理无界流的应用。有界流则由一些专为固定大小数据集特殊设计的算法和数据结构进行内部处理，产生了出色的性能。

1.1.2 部署应用到任意地方

Apache Flink 是一个分布式系统，它需要计算资源来执行应用程序。

Flink集成了所有常见的集群资源管理器，例如 Hadoop YARN、 Apache Mesos 和 Kubernetes，但同时也可以作为独立集群运行（standalone）。

Flink 被设计为能够很好地工作在上述每个资源管理器中，这是通过资源管理器特定(resourcemanager-specific)的部署模式实现的。Flink可以采用与当前资源管理器相适应的方式进行交互。

部署 Flink 应用程序时，Flink会根据应用程序配置的并行性自动标识所需的资源，并从资源管理器请求这些资源。在发生故障的情况下，Flink通过请求新资源来替换发生故障的容器。提交或控制应用程序的所有通信都是通过 REST调用进行的，这可以简化Flink与各种环境中的集成。

1.1.3 运行任意规模应用

Flink 旨在任意规模上运行有状态流式应用。因此，应用程序被并行化为可能数千个任务，这些任务分布在集群中并发执行。所以应用程序能够充分利用无尽的 CPU、内存、磁盘和网络 IO。而且 Flink 很容易维护非常大的应用程序状态。其异步和增量的检查点算法对处理延迟产生最小的影响，同时保证精确一次状态的一致性。

Flink 用户报告了其生产环境中一些令人印象深刻的扩展性数字：

• 每天处理数万亿的事件
• 可以维护几TB大小的状态
• 可以部署上千个节点的集群

1.1.4 利用内存性能

有状态的 Flink 程序针对本地状态访问进行了优化。任务的状态始终保留在内存中，如果状态大小超过可用内存，则会保存在能高效访问的磁盘数据结构中。任务通过访问本地（通常在内存中）状态来进行所有的计算，从而产生非常低的处理延迟。Flink通过定期和异步地对本地状态进行持久化存储来保证故障场景下精确一次的状态一致性。

1.2 Flink架构图

1.3 入门案例演示 1.3.1 pom文件

1. 官网建议使用IDEA，集成Scala和Maven比较方便
2. pom.xml文件指定

    1.10.1
    2.11.8
    1.8
    1.8



    
        org.apache.flink
        flink-streaming-java_2.11
        ${flink.version}
    

1.3.2 单词计数案例演示

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import scala.Tuple2;

public class WordCount {
    public static void main(String[] args) throws Exception {
        // 1. 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 2. 获取数据源
        DataStreamSource dataStream = env.socketTextStream("192.168.8.110", 9999);
        // 3. 数据处理
        SingleOutputStreamOperator> wordAndOne = dataStream.flatMap(new FlatMapFunction>() {
            @Override
            public void flatMap(String line, Collector> collector) throws Exception {
                String[] fields = line.split(",");
                for (String word : fields) {
                    collector.collect(new Tuple2<>(word, 1));
                }
            }
        });

        SingleOutputStreamOperator> wordCount = wordAndOne.keyBy(0).sum(1);

        // 4. 数据输出
        wordCount.print();
        // 5. 启动任务
        env.execute("word count ....");
    }
}

1.3.3 使用面向对象

把数据看成对象，遇到字段较多的数据操作比较方便

public class WordCount {
	public static void main(String[] args) throws Exception {
		// 1 获取执行环境
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		// 2 获取数据源
		DataStreamSource dataStream = env.socketTextStream("192.168.152.102", 9999);
		// 3 数据处理
		SingleOutputStreamOperator wordCount =  dataStream.flatMap(new FlatMapFunction() {
			@Override
			public void flatMap(String line, Collector collector) throws Exception {
				String[] fields = line.split(",");
				for (String word : fields) {
					collector.collect(new WordAndCount(word, 1));
				}
			}
		}).keyBy("word")
		.sum("count");
		
		// 4 数据输出
		wordCount.print();
		// 5 启动任务
		env.execute("word count ...");
	}
	
	public class WordAndCount {
		private String word;
		private int count;

		// getter, setter, toString, 无参构造，有参构造
	}
}

1. 先使用命令启动 netcat网络服务端
   • nc -l -p 9999
2. 再启动代码

1.3.4 使用最佳实践

flink建议如果程序中需要传入参数，使用它提供的ParameterTool

public class WordCount {
    public static void main(String[] args) throws Exception {
        // 1. 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 2. 获取数据源
        // 使用 flink提供的工具类，获取传递的参数
        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        String hostname = parameterTool.get("hostname");
        int port = parameterTool.getInt("port");
        DataStreamSource dataStream = env.socketTextStream(hostname, port);
        
        // 3. 数据处理
        // ...
    }
}

1.3.5 抽离业务功能

工作中开发复杂功能模块，习惯把业务算子抽离出来单独开发，这样代码结构会比较清晰

/// 分割单词
public class SplitWord implements FlatMapFunction {
    @Override
    public void flatMap(String line, Collector collector) throws Exception {
        String[] fields = line.split(",");
        for (String word : fields) {
            collector.collect(new WordAndCount(word, 1));
        }
    }
}

public class WordCount {
    public static void main(String[] args) throws Exception {
        // 1. 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 2. 获取数据源
        // 使用 flink提供的工具类，获取传递的参数
        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        String hostname = parameterTool.get("hostname");
        int port = parameterTool.getInt("port");
        DataStreamSource dataStream = env.socketTextStream(hostname, port);

        // 3. 数据处理
        SingleOutputStreamOperator wordCount = dataStream
                        .flatMap(new SplitWord())
                        .keyBy("word")
                        .sum("count");

        // 4. 数据输出
        wordCount.print();
        // 5. 启动任务
        env.execute("word count ....");
    }
}

1.4 Flink核心概念 1.4.1 Flink概念

Apache Flink is a framework and distributed processing engine for stateful computations over unbounded and bounded data streams

1. 核心概念之stateful

• flink: 有状态的流
• SparkStreaming: 没有状态
  • 2秒运行一次： hadoop,hadoop,hive --> hadoop2,hive1
  • 2秒运行一次: hadoop,hive,hive --> hadoop1,hive2
  • Sparktreaming 有2个算子是有状态的
    • mapWithState, updateStateByKey
    • 里面的状态使用起来不是很灵活

2. 核心概念之Operator
   • source
   • transfrom
   • sink

3. 核心概念之distributed

• 每个算子都可以并行

1.4.2 本地观察Flink任务

1. pom.xml中添加如下依赖

    org.apache.flink
    flink-runtime-web_2.11
    ${flink.version}

2. 修改获取执行环境的代码

StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

3. 运行程序, 程序运行起来打开 http://localhost:8081/

为什么没有设置并行度，wordcount的并行度是8？

• 电脑cpu是8核，所以并行度是8
• source是socket, 只有1个端口。所以source的并行度是1。

1.4.3 并行度

1. 设置全局并行度为2

env.setParallelism(2);

2. 设置sink Operator并行度

wordCount.print().setParallelism(1);

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Gygeg7eJ-1633917012318)(assets/C1142570B79F477D832DC652CEB38CCC.png)]

3. Flink架构

Flink的架构是主从式的架构， 主节点叫：JobManager，从节点叫 TaskManager 。

3个节点部署Flink集群。

4. 任务分布式运行

5. Flink分布式任务调度

1.4.4 数据传输策略

1. forward strategy

1. 一个 task 的输出只发送给一个 task 作为输入；
2. 如果两个 task 都在一个 JVM 中的话，那么就可以避免网络开销；

---

2. key based strategy

1. 数据需要按照某个属性(key)进行分组(或者说分区)；
2. 相同 key 的数据需要传输给同一个 task，在一个 task 中进行处理；

---

3. broadcast strategy

4. random strategy

1. 数据随机的从一个 task 中传输给下一个 operator 所有的 subtask；
2. 保证数据能均匀的传输给所有的 subtask；

TaskManager的并行度 与 Task的数据传播策略的关系

1.4.5 Operator Chain

1. Flink与Kafka版本整合

添加如下依赖：

	org.apache.flink
	flink-connector-kafka-0.11_2.11
	${flink.version}

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

        String topic = "testSlot";
        Properties comsumerProperties = new Properties();
        comsumerProperties.setProperty("bootstrap.servers","192.168.15.102:9092");
        comsumerProperties.setProperty("group.id","testSlot_consumer");

        FlinkKafkaConsumer011 myConsumer =
                new FlinkKafkaConsumer011<>(topic, new SimpleStringSchema(), comsumerProperties);

        DataStreamSource data = env.addSource(myConsumer).setParallelism(3);

        SingleOutputStreamOperator wordoneStream = data.flatMap(new SplitWord()).setParallelism(2) ;

        SingleOutputStreamOperator result = wordOneStream.keyBy(0).sum(1).setParallelism(2);

        result.map(tuple -> tuple.toString()).setParallelism(2).print().setParallelism(1);

        env.execute("wordCount2");
    }
}

2. dataflow效果图如下

注：演示资源不足，修改资源以后，任务正常运行

3. Operator Chain

Operator Chain的条件：

1. 数据传输策略是 forward strategy
2. 在同一个 TaskManager 中运行

并行度都设置为1,观察情况

并行度设置为2，观察情况

并行度设置为3，观察情况

1.5 Flink分布式运行环境 1.5.1 flink分布式四层模型

Flink 代码开发就是要构建一个 dataflow，这个 dataflow 运行需要经历如下 4 个阶段：

1. Stream Graph
2. Job Graph
3. Execution Graph
4. Physical Execution Graph

1.5.2 Flink任务分布式运行流程

二、DataStream API 2.1 Flink之数据源 2.1.1 source简介

source是程序的数据源输入，你可以通过StreamExecutionEnvironment.addSource(sourceFunction)来为你的程序添加一个source。

flink提供了大量的已经实现好的source方法，也可以自定义source：

1. 通过实现sourceFunction接口来自定义无并行度的source
2. 通过实现ParallelSourceFunction 接口 or 继承RichParallelSourceFunction来自定义有并行度的source

大多数情况下，我们使用自带的source即可。

获取source的方式（自带的）

1. 基于文件   - `readTextFile(path)`   - 读取文本文件，文件遵循`TextInputFormat`读取规则，逐行读取并返回。2. 基于socket   - `socketTextStream`   - 从socker中读取数据，元素可以通过一个分隔符切开。3. 基于集合   - `fromCollection(Collection)`   - 通过java 的`collection`集合创建一个数据流，集合中的所有元素必须是相同类型的。4. 扩展数据源   - `addSource` 可以实现读取第三方数据源的数据   - 系统内置提供了一批`connectors`，连接器会提供对应的source支持【kafka】

扩展的数据源

• Apache Kafka (source/sink) 后面重点分析
• Apache Cassandra (sink)
• Amazon Kinesis Streams (source/sink)
• Elasticsearch (sink)
• Hadoop FileSystem (sink)
• RabbitMQ (source/sink)
• Apache NiFi (source/sink)
• Twitter Streaming API (source)

2.1.2 数据源之collection

public class StreamingSourceFromCollection {
    public static void main(String[] args) throws Exception {
        // 1. 获取环境变量
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2. 模拟数据
        ArrayList data = new ArrayList<>();
        data.add("hadoop");
        data.add("spark");
        data.add("flink");

        // 3. 获取数据源
        DataStreamSource dataStream = env.fromCollection(data);
        // 4. transfrom action操作
        SingleOutputStreamOperator addPreStream =
                dataStream.map(new MapFunction() {
            @Override
            public String map(String word) throws Exception {
                return "kaikeba_" + word;
            }
        });

        // 5. 对结果进行处理（打印到控制台）
        addPreStream.print().setParallelism(1);
        // 6. 启动程序
        env.execute("StreamingSourceFromCollection");
    }
}

2.1.3 自定义单并行度数据源

public class MyNoParalleSource implements SourceFunction {
    private long number = 1l;
    private boolean isRunning = true;

    @Override
    public void run(SourceContext sct) throws Exception {
        while(isRunning){
            sct.collect(number);
            number++;
            // 每秒生成一条数据
            Thread.sleep(1000);
        }
    }

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

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

        DataStreamSource numberStream = env.addSource(new MyNoParalleSource()).setParallelism(1);

        SingleOutputStreamOperator dataStream =
                numberStream.map(new MapFunction() {
            @Override
            public Long map(Long value) throws Exception {
                System.out.println("接收到了数据：" + value);
                return value;
            }
        });

        SingleOutputStreamOperator filterDataStream =
                dataStream.filter(new FilterFunction() {
            @Override
            public boolean filter(Long number) throws Exception {
                return number % 2 == 0;
            }
        });
        
        filterDataStream.print().setParallelism(1);
        env.execute("StreamingDemoWithMyNoPralalleSource");
    }
}

运行结果：

2.1.4 自定义多并行度数据源

public class MyParalleSource implements ParallelSourceFunction {
    private long number = 1l;
    private boolean isRunning = true;

    @Override
    public void run(SourceContext sct) throws Exception {
        while(isRunning){
            sct.collect(number);
            number++;
            // 每秒生成一条数据
            Thread.sleep(1000);
        }
    }

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

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

        // 修改数据源，设置并行度
        DataStreamSource numberStream = env.addSource(new MyParalleSource()).setParallelism(2);

        SingleOutputStreamOperator dataStream =
                numberStream.map(new MapFunction() {
                    @Override
                    public Long map(Long value) throws Exception {
                        System.out.println("接收到了数据：" + value);
                        return value;
                    }
                });

        SingleOutputStreamOperator filterDataStream =
                dataStream.filter(new FilterFunction() {
                    @Override
                    public boolean filter(Long number) throws Exception {
                        return number % 2 == 0;
                    }
                });

        filterDataStream.print().setParallelism(1);
        env.execute("StreamingDemoWithMyParalalleSource");
    }
}

运行结果：

2.2 常见Transformation操作 2.2.1 map和filter

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

        DataStreamSource numberStream = env.addSource(new MyNoParalleSource()).setParallelism(1);

        SingleOutputStreamOperator dataStream =
                numberStream.map(new MapFunction() {
                    @Override
                    public Long map(Long value) throws Exception {
                        System.out.println("接收到了数据：" + value);
                        return value;
                    }
                });

        SingleOutputStreamOperator filterDataStream =
                dataStream.filter(new FilterFunction() {
                    @Override
                    public boolean filter(Long number) throws Exception {
                        return number % 2 == 0;
                    }
                });

        filterDataStream.print().setParallelism(1);
        env.execute("MapDemo");
    }
}

2.2.2 flatMap，keyBy和sum

经过测试，发现问题在于POJO类的定义上。对于要充当key的POJO类，必须满足以下条件：

1. 字段名必须声明为public的；
2. 必须有默认的无参构造器；
3. 所有构造器必须声明为public的。

public class WordCount {
    public String word;
    public long count;

    public WordCount() {
    }

    public WordCount(String word, long count) {
        this.word = word;
        this.count = count;
    }

    @Override
    public String toString() {
        return "WordCount{" +
                "word='" + word + ''' +
                ", count=" + count +
                '}';
    }
}

public class WindowWordCountJava {
    public static void main(String[] args) throws Exception {
        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        int port = parameterTool.getInt("port");
        String hostname = parameterTool.get("hostname");
        String delimiter = "t";

        // 1. 获取flink运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 2. 获取数据源
        DataStreamSource textStream = env.socketTextStream(hostname, port, delimiter);
        // 3. 执行transformation操作
        SingleOutputStreamOperator wordCountStream = textStream.flatMap(new FlatMapFunction() {
            @Override
            public void flatMap(String line, Collector collector) throws Exception {
                String[] fileds = line.split("t");
                for (String word : fileds) {
                    collector.collect(new WordCount(word, 1l));
                }
            }
        }).keyBy("word")
                .timeWindow(Time.seconds(2), Time.seconds(1)) // 每隔1秒计算最近2秒的数据
                .sum("count");

        wordCountStream.print().setParallelism(1);
        // 4. 运行程序
        env.execute("WindowWordCountJava");
    }
}

2.2.3 union

// 合并多个流，新的流会包含所有流中的数据，但是union是一个限制，就是所有合并的流类型必须是一致的
public class UnionDemo {
    public static void main(String[] args) throws Exception {
        // 1. 获取运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 2. 获取数据源
        DataStreamSource text1 = env.addSource(new MyNoParalleSource()).setParallelism(1);
        DataStreamSource text2 = env.addSource(new MyNoParalleSource()).setParallelism(1);
        // 3. 把 text1 和 text2 组装到一起
        DataStream text = text1.union(text2);

        DataStream num = text.map(new MapFunction() {
            @Override
            public Long map(Long value) throws Exception {
                System.out.println("原始接收到数据：" + value);
                return value;
            }
        });

        // 4. 每2秒钟处理一次数据
        DataStream sum = num.timeWindowAll(Time.seconds(2)).sum(0);
        // 5. 打印结果
        sum.print().setParallelism(1);
        env.execute(UnionDemo.class.getSimpleName());
    }
}

2.2.4 connect, conMap和conFlatMap

// 和union类似，但是只能连接两个流，两个流的数据类型可以不同，会对两个流中的数据应用不同的处理方法
public class ConnectionDemo {
    public static void main(String[] args) throws Exception {
        // 获取flink环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 获取数据源
        DataStreamSource text1 = env.addSource(new MyNoParalleSource()).setParallelism(1);
        DataStreamSource text2 = env.addSource(new MyNoParalleSource()).setParallelism(1);

        SingleOutputStreamOperator text2_str = text2.map(new MapFunction() {
            @Override
            public String map(Long value) throws Exception {
                return "str_" + value;
            }
        });

        ConnectedStreams connectStream = text1.connect(text2_str);
        SingleOutputStreamOperator