第1章 简介
在前面的文章中,介绍了Flink相关环境的准备,并且完成了一个简单的Flink开发环境的搭建;本篇文章介绍一个完整的端到端涵盖Flink计算的案例:客户端=>Web API服务=>Kafka=>Flink=>MySQL。本次仍然以Flink Table API/SQL为例,采用docker-compose的方式进行部署。(文章中只给出关键部分代码,完整代码详见后续笔者上传的github)。
第2章 docker-compose
2.1 添加docker-compose.yml文件
version: '2'
services:
jobmanager:
image: zihaodeng/flink:1.11.1
volumes:
- D:/21docker/flinkDeploy:/opt/flinkDeploy
hostname: "jobmanager"
expose:
- "6123"
ports:
- "4000:4000"
command: jobmanager
environment:
- JOB_MANAGER_RPC_ADDRESS=jobmanager
taskmanager:
image: zihaodeng/flink:1.11.1
volumes:
- D:/21docker/flinkDeploy:/opt/flinkDeploy
expose:
- "6121"
- "6122"
depends_on:
- jobmanager
command: taskmanager
links:
- jobmanager:jobmanager
environment:
- JOB_MANAGER_RPC_ADDRESS=jobmanager
zookeeper:
container_name: zookeeper
image: zookeeper:3.6.1
ports:
- "2181:2181"
kafka:
container_name: kafka
image: wurstmeister/kafka:2.12-2.5.0
volumes:
- D:/21docker/var/run/docker.sock:/var/run/docker.sock
ports:
- "9092:9092"
depends_on:
- zookeeper
environment:
#KAFKA_ADVERTISED_HOST_NAME: kafka
HOSTNAME_COMMAND: "route -n | awk '/UG[ \t]/{print $$2}'"
KAFKA_CREATE_TOPICS: "order-log:1:1"
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
#KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://127.0.0.1:9092
#KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092
mysql:
image: mysql:5.7
container_name: mysql
volumes:
- D:/21docker/mysql/data/db:/var/lib/mysql/
- D:/21docker/mysql/mysql-3346.sock:/var/run/mysql.sock
- D:/21docker/mysql/data/conf:/etc/mysql/conf.d
ports:
- 3306:3306
command:
--default-authentication-plugin=mysql_native_password
--lower_case_table_names=1
environment:
MYSQL_ROOT_PASSWORD: 123456
TZ: Asia/Shanghai
2.2 docker-compose启动
$ docker-compose up -d
检查运行情况
这一节将docker-compose准备好,后面的作业环境启动会非常的方便,接下来,开始准备相应的程序。
第3章 创建WebApi项目
3.1 创建WebApi(Restful API)接口项目
采用springboot 快速搭建一个API项目,笔者这里采用Restful Api接口格式; 部分代码如下(完整代码详见笔者github)。
创建一个Post接口,供客户端调用
@RestController
@RequestMapping("/order")
public class OrderController {
@Autowired
private Sender sender;
@PostMapping
public String insertOrder(@RequestBody Order order) {
sender.producerKafka(order);
return "{\"code\":0,\"message\":\"insert success\"}";
}
}
创建Sender类,发送数据至Kafka
public class Sender {
@Autowired
private KafkaTemplate<String,String> kafkaTemplate;
private static Random rand = new Random();
public void producerKafka(Order order){
order.setPayTime(String.valueOf(new Timestamp(System.currentTimeMillis()+ rand.nextInt(100))));//EventTime
kafkaTemplate.send("order-log", JSON.toJSONString(order));
}
}
第4章 创建Flink作业
这里采用Flink Table API/SQL,实现一个tumble窗口计算:将Kafka里读取到的数据,经过窗口计算汇总后,写入MySQL中。
public class Kafka2MysqlByEnd2End {
public static void main(String[] args) throws Exception {
// Kafka source
String sourceSQL="CREATE TABLE order_source (\n" +
" payTime VARCHAR,\n" +
" rt as TO_TIMESTAMP(payTime),\n" +
" orderId BIGINT,\n" +
" goodsId INT,\n" +
" userId INT,\n" +
" amount DECIMAL(23,10),\n" +
" address VARCHAR,\n" +
" WATERMARK FOR rt as rt - INTERVAL '2' SECOND\n" +
" ) WITH (\n" +
" 'connector' = 'kafka-0.11',\n" +
" 'topic'='order-log',\n" +
" 'properties.bootstrap.servers'='kafka:9092',\n" +
" 'format' = 'json',\n" +
" 'scan.startup.mode' = 'latest-offset'\n" +
")";
//Mysql sink
String sinkSQL="CREATE TABLE order_sink (\n" +
" goodsId BIGINT,\n" +
" goodsName VARCHAR,\n" +
" amount DECIMAL(23,10),\n" +
" rowtime TIMESTAMP(3),\n" +
" PRIMARY KEY (goodsId) NOT ENFORCED\n" +
" ) WITH (\n" +
" 'connector' = 'jdbc',\n" +
" 'url' = 'jdbc:mysql://mysql:3306/flinkdb?characterEncoding=utf-8&useSSL=false',\n" +
" 'table-name' = 'good_sale',\n" +
" 'username' = 'root',\n" +
" 'password' = '123456',\n" +
" 'sink.buffer-flush.max-rows' = '1',\n" +
" 'sink.buffer-flush.interval' = '1s'\n" +
")";
// 创建执行环境
EnvironmentSettings settings = EnvironmentSettings
.newInstance()
.useBlinkPlanner()
.inStreamingMode()
.build();
//TableEnvironment tEnv = TableEnvironment.create(settings);
StreamExecutionEnvironment sEnv = StreamExecutionEnvironment.getExecutionEnvironment();
//sEnv.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.of(1, TimeUnit.SECONDS)));
//sEnv.enableCheckpointing(1000);
//sEnv.setStateBackend(new FsStateBackend("file:///tmp/chkdir",false));
StreamTableEnvironment tEnv= StreamTableEnvironment.create(sEnv,settings);
Configuration configuration = tEnv.getConfig().getConfiguration();
//设置并行度为1
configuration.set(CoreOptions.DEFAULT_PARALLELISM, 1);
//注册souuce
tEnv.executeSql(sourceSQL);
//注册sink
tEnv.executeSql(sinkSQL);
//UDF 在作业中定义UDF
tEnv.createFunction("exchangeGoods", ExchangeGoodsName.class);
String strSQL=" SELECT " +
" goodsId," +
" exchangeGoods(goodsId) as goodsName, " +
" sum(amount) as amount, " +
" tumble_start(rt, interval '5' seconds) as rowtime " +
" FROM order_source " +
" GROUP BY tumble(rt, interval '5' seconds),goodsId";
//查询数据 插入数据
tEnv.sqlQuery(strSQL).executeInsert("order_sink");
//执行作业
tEnv.execute("统计各商品营业额");
}
}
第5章 创建MySQL库表
5.1 建库
mysql> create database flinkdb;
5.2 建表
mysql> create table good_sale(goodsId bigint primary key, goodsName varchar(100) CHARACTER SET utf8 COLLATE utf8_general_ci, amount decimal(23,10), rowtime timestamp);
注意:主键primary key与flink作业ddl定义的主键相同,flink ddl中定义了主键,connector为upsert模式(Flink将根据主键来决定是插入新行还是更新现有行,保证幂等性),如果没有定义则为append模式(Insert插入模式,主键如果出现冲突,则插入将出现失败)。
到此,环境和代码都已准备完毕,接下来开始运行验证。
第6章 运行作业
6.1 本地调试验证
6.1.1 启动集群
用第2章的方式,启动我们准备好的集群环境:
$ docker-compose up -d
6.1.2 启动作业
本地IDEA中启动WebApi项目和Flink作业项目。
6.1.3 发起请求
这里采用一个http工具模拟客户端发起请求,发送json格式的数据:
6.1.4 查看flink作业运行情况
直接在IDEA控制台查看日志
6.1.5 查看结果
到MySQL中查看结果
到这一步,可以看到,通过本地IDE调试的作业,已经顺利将数据从源头拉取,经过flink的计算,将数据写入到了MySQL。接下来我们将作业完整的提交到集群上运行。
6.2 集群运行验证
6.2.1 打包
打包WebAPI项目jar包
到lotemall-webapi(笔者的WebApi项目)目录下执行打包命令
mvn clean package -DskipTests
准备好Dockerfile文件,和打好的jar包放在同一个目录,运行如下命令,创建Images
$ docker build -t lotemall-webapi .
打包Flink作业jar包
到flink-kafka2mysql(笔者的Flink作业项目)目录下执行打包命令
注意因为我们提交到容器内使用,所以source和sink连接的配置IP要由原先的localhost改为容器名
mvn clean package -DskipTests
将打包好的jar包放到docker的挂载目录
6.2.2 启动集群
用第2章的方式,启动我们准备好的集群环境:
$ docker-compose up -d
6.2.3 启动WebAPI项目
运行docker run命令,启动WebAPI项目
$ docker run --link kafka:kafka --net flink-online_default -e TZ=Asia/Shanghai -d -p 8090:8080 lotemall-webapi
6.2.4 运行Flink作业
进入到flink jobmanager容器内,运行作业
$ docker exec -it flink-online_jobmanager_1 /bin/bash
$ bin/flink run -c sql.Kafka2MysqlByEnd2End /opt/flinkDeploy/flink-kafka2mysql-0.0.1.jar -d
或者通过web界面提交作业jar包
![](https://img-blog.csdnimg.cn/20201018180033421.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2R6aDI4NDYxNjE3Mg==,size_16,color_FFFFFF,t_70)
作业提交后,查看Flink web界面,可以看到我们提交的作业已经开始运行
6.2.5 发起请求
同样采用http工具模拟客户端发起请求,发送json格式的数据:
6.2.6 查看Flink作业运行情况
提交到集群中运行,可以直接在Flink web界面查看作业运行情况
查看生成的Watermark
6.2.7 查看结果
到MySQL中查看结果
第一行goodsName为“蛋糕”的记录,就是本次集群上Flink计算的结果。
总结,本篇文章介绍了采用docker-compose的方式方便的部署集群,并且完成一次完整的Flink流计算案例,但是结果呈现方面略显不足,目前的结果只是导入了MySQL数据库。在后续的文章中,我们采用ElasticSearch与Kibana,将结果呈现成一个漂亮的图形界面。