快速入门Flink (7) —— 小白都喜欢看的Flink流处理之DataSources和DataSinks

不知不觉，这已经是快速入门Flink系列的第7篇博客了。早在第4篇博客中，博主就已经为大家介绍了在批处理中，数据输入Data Sources 与数据输出Data Sinks的各种分类(传送门:Flink批处理的DataSources和DataSinks)。但是大家是否还记得Flink的概念？Flink是 分布式、高性能、随时可用以及准确的为流处理应用程序打造的开源流处理框架。所以光介绍了批处理哪里行呢！本篇博客，我们就来学习Flink流处理的DataSources和DataSinks~

文章目录

1、DataStream API开发
- 1.1 入门案例
- - 1.1.1 Flink流处理程序的一般流程
  - 1.1.2 示例
  - 1.1.3 步骤
  - 1.1.4 参考代码
- 1.2 输入数据集 Data Sources
- - 1.2.1 Flink 在流处理上常见的 Source
  - 1.2.2 基于集合的 source
  - 1.2.3 基于文件的 source
  - 1.2.4 基于网络套接字的 source
  - 1.2.5 自定义的 source
  - - 1.2.5.1 SourceFunction：创建非并行数据源
    - 1.2.5.2 ParallelSourceFunction：创建并行数据源
    - 1.2.5.3 RichParallelSourceFunction：创建并行数据源
  - 1.2.6 基于 kafka 的 source
  - 1.2.7 基于 mysql 的 source 操作
- 1.3 数据输出 Data Sinks
- - 1.3.1 sink 到 kafka
  - 1.3.2 sink 到 mysql
- 结语

1、DataStream API开发

1.1 入门案例

1.1.1 Flink流处理程序的一般流程

1) 获取 Flink 流处理执行环境

2) 构建 source

3) 数据处理

4) 构建 sink

1.1.2 示例

编写 Flink 程序，用来统计单词的数量。

1.1.3 步骤

1) 获取 Flink 批处理运行环境

2) 构建一个 socket 源

3) 使用 flink 操作进行单词统计

4) 打印

说明：如果 linux 上没有安装 nc 服务 ,使用 yum 安装

yum install -y nc

1.1.4 参考代码

import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow/** @Author: Alice菌* @Date: 2020/7/9 08:40* @Description: */
// 入门案例，单词统计
object StreamWordCount {def main(args: Array[String]): Unit = {// 1、 创建流处理的执行环境val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment// 2、 构建数据源，使用的socketval socketDataStream: DataStream[String] = env.socketTextStream("node01",9999, 0)// 3、 数据的处理val wordDataStream: DataStream[(String, Int)] = socketDataStream.flatMap(_.split(" ")).map(_ -> 1)//4. 使用keyBy 进行分流（分组）// 在批处理中针对于dataset， 如果分组需要使用groupby// 在流处理中针对于datastream， 如果分组（分流）使用keyByval groupedDataStream: KeyedStream[(String, Int), Tuple] = wordDataStream.keyBy(0)//5. 使用timeWinodw 指定窗口的长度（每5秒计算一次）// spark-》reduceBykeyAndWindowval windowDataStream: WindowedStream[(String,Int),Tuple,TimeWindow]= groupedDataStream.timeWindow(Time.seconds(5))//6. 使用sum执行累加val sumDataStream: DataStream[(String, Int)] = windowDataStream.sum(1)sumDataStream.print()env.execute("StreamWordCount")}
}

我们来测试下效果如何~

首先我们在linux上开启9999端口

nc -lk 9999

然后我们启动我们的程序，发现也是毫无波澜。

接下来就是见证奇迹的时候了，当我以飞快的速度在命令行中敲下这些字母

然后观察程序的控制台，发现打印出了每5秒内，所有的字符数的个数

有朋友肯定会好奇，为什么scala一次显示为3次，后面只显示了1次