Flink编程基本步骤:

1.创建流执行环境 StreamExecutionEnviroment.getExecutionEnviroment() 获取流环境。

2.加载数据源 Source

3.转换操作 Transformation

4.输出出去Sink,落地到其它的数据仓库,直接打印输出.

关于Flink 数据的基本操作 —— 四种分类

1. 单条数据的操作 map filter

2. 多条数据的操作 window

3. 多个流合并成一个流操作 connect union join

4. 将一个流拆分成多个流操作 ，（split 过期），测输出流（OutputTag）output

Flink输入数据源 source

自带预定义Source

• 基于本地集合Source

  • 应用场景，当程序写完之后，测试当前功能是否可用，开发测试用。

  • 分类

    1. 从元素 fromElements

    2. 从集合 fromCollection

    3. 基于Sequence的generateSequence

    4. 基于开始和结束的DataStream ，fromSequence

• 并行度设置

  • 并行度设置方式

    1.设置配置文件 flink-conf.yaml  parallelism.default: 1
    2.在client端提交任务设置并行度  flink run -p 1
    3.在程序中设置全局并行度  env.setParallelism(1)
    4.算子级别的并行度设置  算子.setParallelism(1)
    优先级：
    算子优先级 > 程序中全局并行度 > client提交作业并行度 > 配置文件中的并行度

  • 全局并行度设置

    env.setParallelism(1);

  • 算子并行度设置

    source.print().setParallelism(2);

• 基于文件的Source

  • 批的方式读取文件，只读取一次——readTextFile

  • 流的方式读取文件，实时根据指定周期去监控文件—— readFile

    监控方式 watchType 两种

    1. FileProcessingMode.PROCESS_CONTINUOUSLY 主要用于修改删除操作比较多场景，根据周期持续读取整个文件（破坏仅一次语义特性）

    2. FileProcessingMode.Process_Once 主要用于仅读取一次，读完就 exit

• 基于Socket的Source

  env.socketTextStream("node1", 9999);

自定义Source

• 主要用于指定特定格式的数据源，生成有界或无界数据流

  1. 有界数据流 - 读取 mysql ，读取 文件

  2. 无界数据流 - for 循环生成数据

• 需求 - 通过用户自定义的方式生成数据源。

• 需求案例

  • 自定义实现SourceFunction接口案例 只需要重写 run方法和 cancel方法

package cn.itcast.flink.api;import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;import java.text.SimpleDateFormat;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.TimeUnit;/*** Author itcast* Date 2021/11/29 14:56* Desc - 每1秒钟随机生成一条订单信息(订单ID、用户ID、订单金额、时间戳)* public class Order* String oid;* int uid;* double money;* long timestamp;* String datetime;* 每一秒钟生成一条数据* 打印输出每条数据* 执行流环境*/
public class OrderSource {public static void main(String[] args) throws Exception {//1.创建流执行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//2.设置并行度env.setParallelism(1);//3.获取自定义数据源//实现方式DataStreamSource<Order> source = env.addSource(new OrderEmitSource());//4.打印输出source.printToErr();//5.执行流环境env.execute();}public static class OrderEmitSource implements SourceFunction<Order> {//定义一个标记，用于标识当前持续生成数据private volatile boolean isRunning = true;/*** 实现数据的生成，并将生成的数据输出 ctx 输出** @param ctx* @throws Exception*/@Overridepublic void run(SourceContext<Order> ctx) throws Exception {//定义随机数Random rm = new Random();//时间转换格式工具SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");//死循环，一直生成数据while (isRunning) {//随机数String oid = UUID.randomUUID().toString();//用户id ，随机 0~5 之间值int uid = rm.nextInt(6);//money 0~100之间的double money = rm.nextDouble()*100;//时间戳long timestamp = System.currentTimeMillis();//当前时间String datetime = sdf.format(timestamp);Order order = new Order(oid,uid,money,timestamp,datetime);//收集数据ctx.collect(order);//程序休眠一秒接着执行TimeUnit.SECONDS.sleep(1);}}/*** 用户取消生成的时候，取消生成*/@Overridepublic void cancel() {isRunning = false;}}@Data@AllArgsConstructor@NoArgsConstructorpublic static class Order {private String oid;private int uid;private double money;private long timestamp;private String datetime;}
}

 实现ParallelSourceFunction 接口案例

 并行化生成数据，算子上设置并行度 setParallelism(n)

package cn.itcast.flink.api;import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;import java.text.SimpleDateFormat;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.TimeUnit;/*** Author itcast* Date 2021/11/29 14:56* Desc - 每1秒钟随机生成一条订单信息(订单ID、用户ID、订单金额、时间戳)* public class Order* String oid;* int uid;* double money;* long timestamp;* String datetime;* 每一秒钟生成一条数据* 打印输出每条数据* 执行流环境*/
public class OrderParallelismSource {public static void main(String[] args) throws Exception {//1.创建流执行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//2.设置并行度env.setParallelism(1);//3.获取自定义数据源//实现方式DataStreamSource<Order> source = env.addSource(new OrderEmitSource()).setParallelism(6);//4.打印输出 source.printToErr();//5.执行流环境env.execute();}public static class OrderEmitSource implements ParallelSourceFunction<Order> {//定义一个标记，用于标识当前持续生成数据private volatile boolean isRunning = true;/*** 实现数据的生成，并将生成的数据输出 ctx 输出** @param ctx* @throws Exception*/@Overridepublic void run(SourceContext<Order> ctx) throws Exception {//定义随机数Random rm = new Random();//时间转换格式工具SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");//死循环，一直生成数据while (isRunning) {//随机数String oid = UUID.randomUUID().toString();//用户id ，随机 0~5 之间值int uid = rm.nextInt(6);//money 0~100之间的double money = rm.nextDouble()*100;//时间戳long timestamp = System.currentTimeMillis();//当前时间String datetime = sdf.format(timestamp);Order order = new Order(oid,uid,money,timestamp,datetime);//收集数据ctx.collect(order);//程序休眠一秒接着执行TimeUnit.SECONDS.sleep(5);}}/*** 用户取消生成的时候，取消生成*/@Overridepublic void cancel() {isRunning = false;}}@Data@AllArgsConstructor@NoArgsConstructorpublic static class Order {private String oid;private int uid;private double money;private long timestamp;private String datetime;}
}

实现RichParallelSourceFunction案例

• Rich 是富函数继承了 AbstractRichFunciton，实现了

• 生命周期的 open 和 close 方法

  1. open 方法，用于实现当前生成的初始化条件

  2. close 方法，用于生成数据结束的收尾工作

  3. getRuntimeContext 方法，用于获取当前的程序的上下文对象（参数、环境变量、状态、累加器等）

• 案例 - 从数据库中读取数据

• 1:初始化工作 —— 创建数据库和数据表

# 创建数据库
create database test;
# 使用数据库
use test;
# 创建表和导入数据
DROP TABLE IF EXISTS `user`;
CREATE TABLE `user`  (`id` int(11) NOT NULL,`username` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,`password` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,`name` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Compact;-- ----------------------------
-- Records of user
-- ----------------------------
INSERT INTO `user` VALUES (10, 'dazhuang', '123456', '大壮');
INSERT INTO `user` VALUES (11, 'erya', '123456', '二丫');
INSERT INTO `user` VALUES (12, 'sanpang', '123456', '三胖');SET FOREIGN_KEY_CHECKS = 1;

 2:Flink读取MySQL的数据源

package cn.itcast.flink.source;import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.concurrent.TimeUnit;/*** Author itcast* Date 2022/1/11 16:17* Desc 读取mysql数据表并打印输出* 开发步骤：* 1.创建和准备数据库和数据表  flink* 2.获取流执行环境* 3.设置并行度* 4.添加自定义数据源，从mysql中读取数据，实现 RichSourceFunction ，rich 增强富功能 open close getRuntimeContext* 4.1. open 初始化动作，创建连接，创建 statement ，获取变量* 4.2. run方法 读取数据表中数据并封装成对象* 4.3. close方法 关闭statement和连接* 5. 打印结果输出* 6. 执行流环境*/
public class UserSource {public static void main(String[] args) throws Exception {//2.获取流执行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//3.设置并行度env.setParallelism(1);//4.添加自定义数据源，从mysql中读取数据，实现 RichSourceFunction ，rich 增强富功能 open close getRuntimeContextDataStreamSource<User> source = env.addSource(new RichSourceFunction<User>() {Connection conn = null;Statement statement = null;//标记boolean isRunning = true;/*** 在所有执行source，首先要做的初始化工作* @param parameters* @throws Exception*/@Overridepublic void open(Configuration parameters) throws Exception {//1.设置 driver 驱动Class.forName("com.mysql.jdbc.Driver");//2.获取连接 设置 url 用户名 密码conn = DriverManager.getConnection("jdbc:mysql://node1:3306/flink?useSSL=false","root","123456");//3.创建 statement 基于 sqlstatement = conn.createStatement();}/*** 所有的元素都在这里执行* @param ctx* @throws Exception*/@Overridepublic void run(SourceContext<User> ctx) throws Exception {String sql = "select id,username,password,name from user";while (isRunning) {//1.读取数据 statement.executeQuery 得到 ResultSet 结果集ResultSet rs = statement.executeQuery(sql);//2.遍历 ResultSet 是否有数据 hasNext() = truewhile (rs.next()) {User user = new User();//3.将每条数据 赋值 对象 Userint id = rs.getInt("id");String username = rs.getString("username");String password = rs.getString("password");String name = rs.getString("name");user.setId(id);user.setUsername(username);user.setPassword(password);user.setName(name);//4.将 User 收集 ctx.collect(user)ctx.collect(user);}TimeUnit.MINUTES.sleep(5);}}@Overridepublic void cancel() {//将flag置为 falseisRunning = false;}/*** 所有的元素执行完毕的收尾工作* @throws Exception*/@Overridepublic void close() throws Exception {//关闭 statementif (!statement.isClosed()) {statement.close();}//关闭 connectionif (!conn.isClosed()) {conn.close();}}});//4.1. open 初始化动作，创建连接，创建 statement ，获取变量//4.2. run方法 读取数据表中数据并封装成对象//4.3. close方法 关闭statement和连接//5. 打印结果输出source.printToErr();//6. 执行流环境env.execute();}public static class User {// idprivate int id;// usernameprivate String username;// passwordprivate String password;// nameprivate String name;public int getId() {return id;}public void setId(int id) {this.id = id;}public String getUsername() {return username;}public void setUsername(String username) {this.username = username;}public String getPassword() {return password;}public void setPassword(String password) {this.password = password;}public String getName() {return name;}public void setName(String name) {this.name = name;}@Overridepublic String toString() {return "User{" +"id=" + id +", username='" + username + '\'' +", password='" + password + '\'' +", name='" + name + '\'' +'}';}}
}