兄弟,干了这杯Flink SQL !!!

1.创业的感受

累,累,累,已经连续创业3天,2晚失眠睡不着,中午午休也睡不着,衣带渐宽终不悔,为伊消得人憔悴,哈哈....。但是觉得自己在成长和进步,连续不断的输出,让我不得已不断的学习,不断的输入。有时候会觉得,特别的孤独,特么的说好大家一起开源,一起创业,但是真正行动的时候,都看不到人,感觉一个人走在漆黑的一望无际的沙漠中,但是我还是要选择走下去呀,你想想打铁还需自生硬,如果我自己都不行动,那么整个团队又知道方向在哪里?也没有人告诉我怎么做,也没有人告诉我方向,让我就扑上去了,不问青红皂白的扑上去,手动补一个狗头!!!

2.望梅止渴,画饼充饥

打造一个开源的社区,能够接靠谱的外包,这样回馈关注我的兄弟们,让他们能够买的起房。同时能够做出伟大的开源产品出来,服务这个世界。这就是我的梦想,我在努力中....

3.干了这杯 Flink SQL!!!

1.前置条件

安装Flink 1.10 版本 【切记一定要在这之上的版本】,Flink下载地址

安装kafka

安装zookeeper

安装mysql

2.下载我提供的包

关注微信公众号 【LarkMidTable】,回复【flinksql学习】获取

3.替换Flink的lib包,用我提供的lib包

flinkx-sql/lib 替换 flink-1.10.0/lib

4.mysql中创建库和表

库名: flink-test

表 : pvuv_sink.sql

5.假设你的Flink,Mysql,kafka都启动的情况下copy到Linux服务器

flinkx-sql\target\flink-sql-submit.jar

flinkx-sql\src\main\resources\q1.sql

q1.sql需要修改的 (1.zookeeper的IP,2.kafka的IP,3.Mysql的IP、用户名、密码)

CREATE TABLE user_log (
  user_id VARCHAR,
  item_id VARCHAR,
  category_id VARCHAR,
  behavior VARCHAR,
  ts TIMESTAMP
) WITH (
  'connector.type' = 'kafka',
  'connector.version' = 'universal',
  'connector.topic' = 'user_behavior',
  'connector.startup-mode' = 'earliest-offset',
  'connector.properties.0.key' = 'zookeeper.connect',
  'connector.properties.0.value' = 'localhost:2181',
  'connector.properties.1.key' = 'bootstrap.servers',
  'connector.properties.1.value' = 'localhost:9092',
  'update-mode' = 'append',
  'format.type' = 'json',
  'format.derive-schema' = 'true'
);

-- sink
CREATE TABLE pvuv_sink (
  dt VARCHAR,
  pv BIGINT,
  uv BIGINT
) WITH (
  'connector.type' = 'jdbc',
  'connector.url' = 'jdbc:mysql://localhost:3306/flink-test',
  'connector.table' = 'pvuv_sink',
  'connector.username' = 'root',
  'connector.password' = '123456',
  'connector.write.flush.max-rows' = '1'
);


INSERT INTO pvuv_sink
SELECT
DATE_FORMAT(ts, 'yyyy-MM-dd HH:00') dt,
COUNT(*) AS pv,
COUNT(DISTINCT user_id) AS uv
FROM user_log
GROUP BY DATE_FORMAT(ts, 'yyyy-MM-dd HH:00');

6.启动开始运行

1.往kafka里灌数据

java -cp flink-sql-submit.jar com.github.wuchong.sqlsubmit.SourceGenerator 1000 | /home/hadoop/data/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic user_behavior

2.启动flink的任务

${basepath}/flink-1.10.0/bin/flink run -d -p 1 flink-sql-submit.jar -w ${q1.sql所在的目录} -f q1.sql

3.验证结果

查看Flink界面 http://localhost:8081

数据库表中是否有聚合数据

4.我是如何学习的呢?

大佬博客,我在上面整合成单机版本,去掉容器版本,同时我将所有的坑都已经踩了,应该没有啥坑了吧!!!

一起努力,一起加油!

上一篇:Python MySQL - mysql-connector 驱动


下一篇:MySQL8计划任务