本文就flink流计算的任务工程模板做一介绍，初始数据源和计算结果数据存储都是通过JDBC连接到数据库。本文以mysql为数据库进行演示。
依赖包
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>8.0.15</version>
</dependency>
工程结构说明
工程结构
本工程包含以下几个部分：
- 结构化数据定义（DTO），本文以常见的学生基本信息为示例。
package com.crazyice.lee.data;
import lombok.*;
@Data
@AllArgsConstructor
@NoArgsConstructor
@ToString(exclude = {"password","age"})
@RequiredArgsConstructor()
public class Student {
private int studentId;
private String name;
private String password;
private boolean sex;
private int age;
}
- 任务编排，这里演示了一个完整流任务的各个关键环节，包括：环境配置、源数据读取（DataStreamSource）、处理过程（数据过滤筛选）、处理结果输出入库（DataSink）、启动任务执行等。
package com.crazyice.lee.jobs;
import com.crazyice.lee.data.Student;
import com.crazyice.lee.reader.JdbcReader;
import com.crazyice.lee.writer.JdbcWriter;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@Slf4j
public class Mysql2MysqlJob {
public static void main(String[] args) throws Exception {
//设置环境
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//获取数据-数据源
DataStreamSource<Student> students = env.addSource(new JdbcReader());
students.name("从mysql读取数据");
//处理数据过程
DataStream<Student> subStudents = students
.filter(e -> e.isSex()).name("过滤男生")
.filter(e -> e.getAge() > 12).name("过滤12岁以上");
//写入数据-处理结果
subStudents.addSink(new JdbcWriter()).name("写入备份学生库");
env.execute("读取Mysql，数据过滤，写入Mysql");
}
}
- 初始数据源、结果写入数据源配置，这部分通过实现RichSourceFunction、RichSinkFunction接口并配合配置文件实现，是相对固定的模式，如果要提高编程效率，可以考虑使用模板及配置的方式动态实现，因为代码结构固定，所以不在此处粘贴，如果感兴趣可以通过文章下面的开源链接查看。
测试验证
flink支持本地测试模式，可以在编码过程中排查错误，只需要在环境设置部分使用本地环境对象即可（将原先的getExecutionEnvironment()替换为createLocalEnvironment()）也可以使用profile来进行编译配置。
//设置环境
final ExecutionEnvironment environment = properties.getProperty("local.running").equalsIgnoreCase("true") ? StreamExecutionEnvironment.createLocalEnvironment() : StreamExecutionEnvironment.getExecutionEnvironment();
网友评论