01 引言
在前面的博客，我们已经对Flink的原理有了一定的了解了.
本文开始讲解Flink程序模型对应的代码，也就是Flink批流一体对应的API，分别对应为：Source 、Transformation、Slink，本文讲Source。
02 Source
Source对应的就是Flink编程模型里面的Data Source数据源：
在Flink官网，我们可以知道Source有如下几种类型：
转义为中文即：
- File-based：基于文件的的Source
- Socket-based: 基于Socket的Source
- Collection-based: 基于集合的Source
- Custom: 自定义Source
2.1 基于集合的Source
相关API（一般用于学习测试时编造数据时使用）：
- env.fromElements(可变参数)；
- env.fromColletion(各种集合)；
- env.generateSequence(开始,结束)；
- env.fromSequence(开始,结束)。
示例代码：
/**
* 把本地的普通的Java集合/Scala集合变为分布式的Flink的DataStream集合!
*
* @author : YangLinWei
* @createTime: 2022/3/7 2:55 下午
* <p>
* 1.env.fromElements(可变参数);
* 2.env.fromColletion(各种集合);
* 3.env.generateSequence(开始,结束);
* 4.env.fromSequence(开始,结束);
*/
public class SourceDemo1 {
public static void main(String[] args) throws Exception {
//1.env
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
//2.source
// * 1.env.fromElements(可变参数);
DataStream<String> ds1 = env.fromElements("hadoop", "spark", "flink");
// * 2.env.fromColletion(各种集合);
DataStream<String> ds2 = env.fromCollection(Arrays.asList("hadoop", "spark", "flink"));
// * 3.env.generateSequence(开始,结束);
DataStream<Long> ds3 = env.generateSequence(1, 10);
//* 4.env.fromSequence(开始,结束);
DataStream<Long> ds4 = env.fromSequence(1, 10);
//3.Transformation
//4.sink
ds1.print();
ds2.print();
ds3.print();
ds4.print();
//5.execute
env.execute();
}
}
运行结果：
2.2 基于文件的Source
相关API（一般用于学习测试）：
- env.readTextFile（本地/HDFS文件/文件夹/压缩文件)
示例代码：
/**
* env.readTextFile（本地/HDFS文件/文件夹/压缩文件)
*
* @author : YangLinWei
* @createTime: 2022/3/7 2:59 下午
*/
public class SourceDemo2 {
public static void main(String[] args) throws Exception {
//1.env
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
//2.source
// * 1.env.readTextFile(本地文件/HDFS文件);//压缩文件也可以
DataStream<String> ds1 = env.readTextFile("data/input/words.txt");
DataStream<String> ds2 = env.readTextFile("data/input/dir");
DataStream<String> ds3 = env.readTextFile("hdfs://node1:8020//wordcount/input/words.txt");
DataStream<String> ds4 = env.readTextFile("data/input/wordcount.txt.gz");
//3.Transformation
//4.sink
ds1.print();
ds2.print();
ds3.print();
ds4.print();
//5.execute
env.execute();
}
}2.3 基于Socket的Source
需求：在node1上使用nc -lk 9999 向指定端口发送数据（nc是netcat的简称，原本是用来设置路由器，我们可以利用它向某个端口发送数据），如果没有该命令可以下安装：
yum install -y nc
使用Flink编写流处理应用程序实时统计单词数量，代码如下：
/**
* SocketSource
*
* @author : YangLinWei
* @createTime: 2022/3/7 3:02 下午
*/
public class SourceDemo3 {
public static void main(String[] args) throws Exception {
//1.env
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
//2.source
DataStream<String> linesDS = env.socketTextStream("node1", 9999);
//3.处理数据-transformation
//|_____3.1每一行数据按照空格切分成一个个的单词组成一个集合
DataStream<String> wordsDS = linesDS.flatMap(new FlatMapFunction<String, String>() {
@Override
public void flatMap(String value, Collector<String> out) throws Exception {
//value就是一行行的数据
String[] words = value.split(" ");
for (String word : words) {
out.collect(word);//将切割处理的一个个的单词收集起来并返回
}
}
});
//|_____3.2对集合中的每个单词记为1
DataStream<Tuple2<String, Integer>> wordAndOnesDS = wordsDS.map(new MapFunction<String, Tuple2<String, Integer>>() {
@Override
public Tuple2<String, Integer> map(String value) throws Exception {
//value就是进来一个个的单词
return Tuple2.of(value, 1);
}
});
//|_____3.3对数据按照单词(key)进行分组
//KeyedStream<Tuple2<String, Integer>, Tuple> groupedDS = wordAndOnesDS.keyBy(0);
KeyedStream<Tuple2<String, Integer>, String> groupedDS = wordAndOnesDS.keyBy(t -> t.f0);
//|_____3.4对各个组内的数据按照数量(value)进行聚合就是求sum
DataStream<Tuple2<String, Integer>> result = groupedDS.sum(1);
//4.输出结果-sink
result.print();
//5.触发执行-execute
env.execute();
}
}
2.4 自定义Source
2.4.1 案例 - 随机生成数据
Flink还提供了数据源接口，我们实现该接口就可以实现自定义数据源，不同的接口有不同的功能，分类如下：
- SourceFunction：非并行数据源(并行度只能=1)
- RichSourceFunction：多功能非并行数据源(并行度只能=1)
- ParallelSourceFunction：并行数据源(并行度能够>`=1)
- RichParallelSourceFunction：多功能并行数据源(并行度能够>`=1) ，Kafka数据源使用的就是该接口。
需求：每隔1秒随机生成一条订单信息(订单ID、用户ID、订单金额、时间戳)
要求:
- 随机生成订单ID(UUID)
- 随机生成用户ID(0-2)
- 随机生成订单金额(0-100)
- 时间戳为当前系统时间
示例代码：
/**
* 自定义Source
*
* @author : YangLinWei
* @createTime: 2022/3/7 3:08 下午
* Flink还提供了数据源接口,我们实现该接口就可以实现自定义数据源，不同的接口有不同的功能，分类如下：
* SourceFunction:非并行数据源(并行度只能=1)
* RichSourceFunction:多功能非并行数据源(并行度只能=1)
* ParallelSourceFunction:并行数据源(并行度能够>=1)
* RichParallelSourceFunction:多功能并行数据源(并行度能够>=1)--后续学习的Kafka数据源使用的就是该接口
*/
public class SourceDemo4 {
public static void main(String[] args) throws Exception {
//1.env
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
//2.Source
DataStream<Order> orderDS = env
.addSource(new MyOrderSource())
.setParallelism(2);
//3.Transformation
//4.Sink
orderDS.print();
//5.execute
env.execute();
}
@Data
@NoArgsConstructor
@AllArgsConstructor
public static class Order {
private String id;
private Integer userId;
private Integer money;
private Long createTime;
}
public static class MyOrderSource extends RichParallelSourceFunction<Order> {
private Boolean flag = true;
@Override
public void run(SourceContext<Order> ctx) throws Exception {
Random random = new Random();
while (flag) {
Thread.sleep(1000);
String id = UUID.randomUUID().toString();
int userId = random.nextInt(3);
int money = random.nextInt(101);
long createTime = System.currentTimeMillis();
ctx.collect(new Order(id, userId, money, createTime));
}
}
//取消任务/执行cancle命令的时候执行
@Override
public void cancel() {
flag = false;
}
}
}
运行结果如下：
2.4.2 案例 - MySQL
需求：实际开发中，经常会实时接收一些数据，要和MySQL中存储的一些规则进行匹配,那么这时候就可以使用Flink自定义数据源从MySQL中读取数据。
那么现在先完成一个简单的需求:
- 从MySQL中实时加载数据；
- 要求MySQL中的数据有变化,也能被实时加载出来。
首先准备数据：
CREATE TABLE t_student (
id int(11) NOT NULL AUTO_INCREMENT,
name varchar(255) DEFAULT NULL,
age int(11) DEFAULT NULL,
PRIMARY KEY (id)
) ENGINE=InnoDB AUTO_INCREMENT=7 DEFAULT CHARSET=utf8;
INSERT INTO t_student VALUES ('1', 'jack', '18');
INSERT INTO t_student VALUES ('2', 'tom', '19');
INSERT INTO t_student VALUES ('3', 'rose', '20');
INSERT INTO t_student VALUES ('4', 'tom', '19');
INSERT INTO t_student VALUES ('5', 'jack', '18');
INSERT INTO t_student VALUES ('6', 'rose', '20');
代码实现如下：
/**
* 简单的需求:
* 从MySQL中实时加载数据
* 要求MySQL中的数据有变化,也能被实时加载出来
*
* @author : YangLinWei
* @createTime: 2022/3/7 3:17 下午
*/
public class SourceDemo5 {
public static void main(String[] args) throws Exception {
//1.env
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//2.Source
DataStream<Student> studentDS = env.addSource(new MySQLSource()).setParallelism(1);
//3.Transformation
//4.Sink
studentDS.print();
//5.execute
env.execute();
}
@Data
@NoArgsConstructor
@AllArgsConstructor
public static class Student {
private Integer id;
private String name;
private Integer age;
}
public static class MySQLSource extends RichParallelSourceFunction<Student> {
private Connection conn = null;
private PreparedStatement ps = null;
@Override
public void open(Configuration parameters) throws Exception {
//加载驱动,开启连接
//Class.forName("com.mysql.jdbc.Driver");
conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/big_data", "root", "123456");
String sql = "select id,name,age from t_student";
ps = conn.prepareStatement(sql);
}
private boolean flag = true;
@Override
public void run(SourceContext<Student> ctx) throws Exception {
while (flag) {
ResultSet rs = ps.executeQuery();
while (rs.next()) {
int id = rs.getInt("id");
String name = rs.getString("name");
int age = rs.getInt("age");
ctx.collect(new Student(id, name, age));
}
TimeUnit.SECONDS.sleep(5);
}
}
@Override
public void cancel() {
flag = false;
}
@Override
public void close() throws Exception {
if (conn != null) conn.close();
if (ps != null) ps.close();
}
}
}
运行结果：
03 文末
本文主要讲解Flink批流一体API中的Source用法，谢谢大家的阅读，本文完！