Flink 使用(一)——从kafka中读取数据写入到HBASE中

Stella981
• 阅读 958

1、前言

  本文是在《如何计算实时热门商品》[1]一文上做的扩展,仅在功能上验证了利用Flink消费Kafka数据,把处理后的数据写入到HBase的流程,其具体性能未做调优。此外,文中并未就Flink处理逻辑做过多的分析,只因引文(若不特殊说明,文中引文皆指《如何计算实时热门商品》一文)中写的很详细了,故仅给出博主调试犯下的错。文中若有错误,欢迎大伙留言指出,谢谢

  源码在GitHub上,地址:https://github.com/L-Wg/flinkExample

  环境:Flink 1.6+Kafka 1.1+HBase 1.2

       OpenJDK 1.8+Maven 3.5.2

2、获取数据

  本文是将Kafka作为数据源(目前业界比较流行的做法),数据的格式和引文的格式一致,数据类型为POJO。为添加源,一般是实现接口SourceFunction,但是Flink与Kafka的链接器(connector),Flink社区已经做好了,我们只需在pom文件中加入相应的依赖即可。这里有值得注意的一点是:flink-connector-kafka-*.jar是有版本要求的,其具体的要求可以参加Flink官网connector一节**[2]**。代码如下:

DataStream<UserBehaviorSchema> dataStream=env.addSource(new FlinkKafkaConsumer010<UserBehaviorSchema>(
                topic,
                new UserBehaviorSerial(),
                properties
        ).setStartFromEarliest());

 其中,在代码中需指定的有:要消费的topic、数据序列化的对象以及配置,其中,配置可指定bootstrap.servers即可,其他配置按需设置。调用setStarFromEarliest()是为让Flink从头消费指定topic中数据,这样写的好处是:只要你Kafka topic中存在数据,测试时就不用重新往kafka里写数据了。当然调用该方法不仅仅是这个作用,其在业务上的使用需根据需求。此外,Flink中还有诸多指定消费kafka的方法,详情请见官网**[2]**。

这里值得说的一点是获取数据后,dataStream的值是不变的,不会因为做过flatmap等操作后就会改变。

3、数据转换

  对Flink 代码的分析过程见引文,此处仅有以下几点需说明的:

  1.  若是kafka中的数据是自己按照因为数据格式随机生成的,请不要按照博主代码中customWaterExtractor()类的写法去定义watermark和timestamp,因为代码中的currentTimeStamp的值可能也是随机的,所以就会造成程序不报错但是卡死等待的情况。

  2.  timestamp的值要和数据源中数据保持相同的数据级。

public static class customWaterExtractor implements AssignerWithPeriodicWatermarks<UserBehaviorSchema>{

        private static final long serialVersionUID = 298015256202705122L;

        private final long maxOutOrderness=3500;
        private long currentTimeStamp=Long.MIN_VALUE;

        @Nullable
        @Override
        public Watermark getCurrentWatermark() {
            return new Watermark(currentTimeStamp-maxOutOrderness);
        }

        @Override
        public long extractTimestamp(UserBehaviorSchema element, long previousElementTimestamp) {
//          此处需要注意的点:timestamp得到的值是否乘以1000转换为毫秒,需要根据消息中被指定为timestamp字段的单位。
            long timeStamp=element.timestamp*1000;
            currentTimeStamp=Math.max(timeStamp,currentTimeStamp);
            return timeStamp;
        }
    }

3.  在返回的结果类ResultEvent中,使用sinking字段去保存HotTopN的名次,其默认值为0。

4、数据存储

  本文中是通过extends RichSinkFunction来实现将数据写入HBase中,其中,@Override的invoke()方法是针对每条数据都会调用的,其余的open()、close()方法,从日志上看是不是针对每条数据都会调用。对open()方法用于打开链接,最好实现连接池避免链接过多,此处HBase的connection已自身实现不用单独实现。

  数据写入HBase时,有两点建议:

  1.  将数据写入HBase的表中时,最好先做好表的预分区工作,避免后期因为表的split造成性能下降以及维护上的困难;

  2.  为加快HBase的查询速度,可以将制定字段作为HBase表的rowkey,文中是指定时间戳和排名作为表的rowkey,至于二级索引等暂不在此处讨论。

5、参考文献链接:

  [1]http://wuchong.me/blog/2018/11/07/use-flink-calculate-hot-items/

  [2]https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/connectors/kafka.html

点赞
收藏
评论区
推荐文章
blmius blmius
2年前
MySQL:[Err] 1292 - Incorrect datetime value: ‘0000-00-00 00:00:00‘ for column ‘CREATE_TIME‘ at row 1
文章目录问题用navicat导入数据时,报错:原因这是因为当前的MySQL不支持datetime为0的情况。解决修改sql\mode:sql\mode:SQLMode定义了MySQL应支持的SQL语法、数据校验等,这样可以更容易地在不同的环境中使用MySQL。全局s
Jacquelyn38 Jacquelyn38
3年前
2020年前端实用代码段,为你的工作保驾护航
有空的时候,自己总结了几个代码段,在开发中也经常使用,谢谢。1、使用解构获取json数据let jsonData  id: 1,status: "OK",data: 'a', 'b';let  id, status, data: number   jsonData;console.log(id, status, number )
皕杰报表之UUID
​在我们用皕杰报表工具设计填报报表时,如何在新增行里自动增加id呢?能新增整数排序id吗?目前可以在新增行里自动增加id,但只能用uuid函数增加UUID编码,不能新增整数排序id。uuid函数说明:获取一个UUID,可以在填报表中用来创建数据ID语法:uuid()或uuid(sep)参数说明:sep布尔值,生成的uuid中是否包含分隔符'',缺省为
Stella981 Stella981
2年前
Python之time模块的时间戳、时间字符串格式化与转换
Python处理时间和时间戳的内置模块就有time,和datetime两个,本文先说time模块。关于时间戳的几个概念时间戳,根据1970年1月1日00:00:00开始按秒计算的偏移量。时间元组(struct_time),包含9个元素。 time.struct_time(tm_y
Wesley13 Wesley13
2年前
mysql设置时区
mysql设置时区mysql\_query("SETtime\_zone'8:00'")ordie('时区设置失败,请联系管理员!');中国在东8区所以加8方法二:selectcount(user\_id)asdevice,CONVERT\_TZ(FROM\_UNIXTIME(reg\_time),'08:00','0
Stella981 Stella981
2年前
HIVE 时间操作函数
日期函数UNIX时间戳转日期函数: from\_unixtime语法:   from\_unixtime(bigint unixtime\, string format\)返回值: string说明: 转化UNIX时间戳(从19700101 00:00:00 UTC到指定时间的秒数)到当前时区的时间格式举例:hive   selec
Wesley13 Wesley13
2年前
00:Java简单了解
浅谈Java之概述Java是SUN(StanfordUniversityNetwork),斯坦福大学网络公司)1995年推出的一门高级编程语言。Java是一种面向Internet的编程语言。随着Java技术在web方面的不断成熟,已经成为Web应用程序的首选开发语言。Java是简单易学,完全面向对象,安全可靠,与平台无关的编程语言。
Stella981 Stella981
2年前
Django中Admin中的一些参数配置
设置在列表中显示的字段,id为django模型默认的主键list_display('id','name','sex','profession','email','qq','phone','status','create_time')设置在列表可编辑字段list_editable
Wesley13 Wesley13
2年前
MySQL部分从库上面因为大量的临时表tmp_table造成慢查询
背景描述Time:20190124T00:08:14.70572408:00User@Host:@Id:Schema:sentrymetaLast_errno:0Killed:0Query_time:0.315758Lock_
Python进阶者 Python进阶者
4个月前
Excel中这日期老是出来00:00:00,怎么用Pandas把这个去除
大家好,我是皮皮。一、前言前几天在Python白银交流群【上海新年人】问了一个Pandas数据筛选的问题。问题如下:这日期老是出来00:00:00,怎么把这个去除。二、实现过程后来【论草莓如何成为冻干莓】给了一个思路和代码如下:pd.toexcel之前把这