Flink on zeppelin 结合kafka实时计算pv uv写入mysql

本文涉及的产品
云数据库 RDS MySQL,集群系列 2核4GB
推荐场景:
搭建个人博客
RDS MySQL Serverless 基础系列,0.5-2RCU 50GB
实时计算 Flink 版,5000CU*H 3个月
简介: 上一篇文章主要介绍了Flink on zeppelin的安装和使用,配置了yarn的模式跑通了一个streaming wordcount的例子,本文主要介绍结合kafka的使用,实时计算一个简单的pv,uv把结果写入到mysql的例子.添加依赖包首先需要添加kafka以及mysql的jar包,有两种方式,第一种是直接把jar包添加到Flink的lib下面,如下所示:

上一篇文章主要介绍了Flink on zeppelin的安装和使用,配置了yarn的模式跑通了一个streaming wordcount的例子,本文主要介绍结合kafka的使用,实时计算一个简单的pv,uv把结果写入到mysql的例子.


添加依赖包


首先需要添加kafka以及mysql的jar包,有两种方式,第一种是直接把jar包添加到Flink的lib下面,如下所示:




只需要添加 flink-sql-connector-kafka_2.11-1.11.0.jar , flink-json-1.11.0.jar , flink-jdbc_2.11-1.10.1.jar , mysql-connector-java-5.1.47.jar 这4个jar包就可以了,我加的比较多是别的地方用到了,用不到的可以不用加防止出现jar包冲突的问题.


第二种是在zeppelin的UI上运行添加依赖包的命令,添加的格式如下所示,


flink.execution.packages  groupId:artifactId:version 然后点击运行就可以了,执行完后需要重启一下 interpreter.


%flink.conf
flink.execution.packages org.apache.flink:flink-jdbc_2.11:1.10.1


flink.execution.packages  这个配置也类似flink.execution.jars,但它不是用来指定jar包,而是用来指定package的。Zeppelin会下载这个package以及这个package的依赖,并且放到flink interpreter的classpath上。如果需要添加多个依赖的话,中间用逗号隔开就可以了.


创建表


先来创建一个kafka的流表,SQL语句如下所示.


%flink.ssql
DROP TABLE IF EXISTS kafka_table;
CREATE TABLE kafka_table (
    name VARCHAR COMMENT '姓名',
    age int COMMENT '年龄',
  city VARCHAR,
    borth VARCHAR,
    ts BIGINT  COMMENT '时间戳',
    t as TO_TIMESTAMP(FROM_UNIXTIME(ts/1000,'yyyy-MM-dd HH:mm:ss')),
    proctime as PROCTIME(),
    WATERMARK FOR t AS t - INTERVAL '5' SECOND
)
WITH (
    'connector' = 'kafka', -- 使用 kafka connector
    'topic' = 'jason_flink',  -- kafka topic
    'scan.startup.mode' = 'latest-offset', -- 从起始 offset 开始读取
    'properties.bootstrap.servers' = 'master:9092,storm1:9092,storm2:9092',  -- broker连接信息
    'properties.group.id' = 'jason_flink_test',
    'scan.startup.mode' = 'latest-offset',  -- 读取数据的位置
    'format' = 'json'  -- 数据源格式为 json
)


这里使用的是Flink1.11.0的版本,所以Connector 的参数个数已经变了,虽然现在也兼容老的写法,不过还是建议使用新版本的写法,这样更加的简洁,然后可以先执行一下查询kafka表的SQL看一下是否可以获取到数据.



数据正常的打印出来了,说明是可以接收到数据的.然后继续创建一个mysql的结果表.


%flink.ssql
drop table if EXISTS  a;
CREATE TABLE a (
  name STRING,
  pv INT not null,
  uv INT not null,
  t_start TIMESTAMP(3),
  t_end TIMESTAMP(3),
  PRIMARY KEY (name) NOT ENFORCED
) WITH (
   'connector' = 'jdbc',
   'url' = 'jdbc:mysql://master:3306/test',
   'table-name' = 'a',
   'username' = 'mysql',
   'password' = '12345678'
)


这里先执行一下show tables也可以看到刚才创建的2个表



执行SQL


然后就可以做一个简单的基于滚动窗口的pv,uv的统计了,SQL语句非常的简单,这里要注意的是query的字段类型要和sink的字段类型保持一致,否则会报字段类型不匹配的错.


%flink.ssql(type=update,parallelism=4)
insert into a 
select name, 
 cast(count(name) as INT) as pv,
 cast(count(distinct name) as INT) as uv,
 TUMBLE_START(t, INTERVAL '5' second) as t_start,
 TUMBLE_END(t, INTERVAL '5' second) as t_end
 from kafka_table 
 group by name,TUMBLE(t, INTERVAL '5' second);


点击右上角的Flink Job,就可以调到Flink的UI页面看到Job运行的情况了.


网络异常,图片无法展示
|


从上面的records received和records send能看到数据进来了.最后再来看一下mysql里面是否有数据.


网络异常,图片无法展示
|


后面会介绍使用Flink on zeppelin实现更多的场景,大家可以持续关注一下.

相关文章
|
11天前
|
运维 数据处理 数据安全/隐私保护
阿里云实时计算Flink版测评报告
该测评报告详细介绍了阿里云实时计算Flink版在用户行为分析与标签画像中的应用实践,展示了其毫秒级的数据处理能力和高效的开发流程。报告还全面评测了该服务在稳定性、性能、开发运维及安全性方面的卓越表现,并对比自建Flink集群的优势。最后,报告评估了其成本效益,强调了其灵活扩展性和高投资回报率,适合各类实时数据处理需求。
|
2月前
|
消息中间件 前端开发 Kafka
【Azure 事件中心】使用Apache Flink 连接 Event Hubs 出错 Kafka error: No resolvable bootstrap urls
【Azure 事件中心】使用Apache Flink 连接 Event Hubs 出错 Kafka error: No resolvable bootstrap urls
|
12天前
|
存储 运维 监控
阿里云实时计算Flink版的评测
阿里云实时计算Flink版的评测
44 15
|
11天前
|
运维 分布式计算 监控
评测报告:阿里云实时计算Flink版
本评测主要针对阿里云实时计算Flink版在用户行为分析中的应用。作为一名数据分析师,我利用该服务处理了大量日志数据,包括用户点击流和登录行为。Flink的强大实时处理能力让我能够迅速洞察用户行为变化,及时调整营销策略。此外,其卓越的性能和稳定性显著降低了运维负担,提升了项目效率。产品文档详尽且易于理解,但建议增加故障排查示例。
|
11天前
|
机器学习/深度学习 运维 监控
阿里云实时计算Flink版体验评测
阿里云实时计算Flink版提供了完善的产品内引导和丰富文档,使初学者也能快速上手。产品界面引导清晰,内置模板简化了流处理任务。官方文档全面,涵盖配置、开发、调优等内容。此外,该产品在数据开发和运维方面表现优秀,支持灵活的作业开发和自动化运维。未来可增强复杂事件处理、实时可视化展示及机器学习支持,进一步提升用户体验。作为阿里云大数据体系的一部分,它能与DataWorks、MaxCompute等产品无缝联动,构建完整的实时数据处理平台。
|
2月前
|
消息中间件 监控 Java
联通实时计算平台问题之监控Kafka集群的断传和积压情况要如何操作
联通实时计算平台问题之监控Kafka集群的断传和积压情况要如何操作
|
2月前
|
消息中间件 监控 Kafka
联通实时计算平台问题之Flink状态后端数据量较大时,问题排查要如何进行
联通实时计算平台问题之Flink状态后端数据量较大时,问题排查要如何进行
|
2月前
|
消息中间件 存储 关系型数据库
实时计算 Flink版产品使用问题之如何使用Kafka Connector将数据写入到Kafka
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
2月前
|
消息中间件 监控 Kafka
实时计算 Flink版产品使用问题之处理Kafka数据顺序时,怎么确保事件的顺序性
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
2月前
|
消息中间件 Java Kafka
【Azure 事件中心】开启 Apache Flink 制造者 Producer 示例代码中的日志输出 (连接 Azure Event Hub Kafka 终结点)
【Azure 事件中心】开启 Apache Flink 制造者 Producer 示例代码中的日志输出 (连接 Azure Event Hub Kafka 终结点)

热门文章

最新文章

下一篇
无影云桌面