生产SparkStreaming数据零丢失最佳实践(含代码)
MySQL创建存储offset的表格
mysql> use test mysql> create table hlw_offset( topic varchar(32), groupid varchar(50), partitions int, fromoffset bigint, untiloffset bigint, primary key(topic,groupid,partitions) );
Maven依赖包
<scala.version>2.11.8</scala.version> <spark.version>2.3.1</spark.version> <scalikejdbc.version>2.5.0</scalikejdbc.version> -------------------------------------------------- <dependency> <groupId>org.scala-lang</groupId> <artifactId>scala-library</artifactId> <version>${scala.version}</version> </dependency> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-core_2.11</artifactId> <version>${spark.version}</version> </dependency> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-sql_2.11</artifactId> <version>${spark.version}</version> </dependency> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-streaming_2.11</artifactId> <version>${spark.version}</version> </dependency> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-streaming-kafka-0-8_2.11</artifactId> <version>${spark.version}</version> </dependency> <dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId> <version>5.1.27</version> </dependency> <!-- https://mvnrepository.com/artifact/org.scalikejdbc/scalikejdbc --> <dependency> <groupId>org.scalikejdbc</groupId> <artifactId>scalikejdbc_2.11</artifactId> <version>2.5.0</version> </dependency> <dependency> <groupId>org.scalikejdbc</groupId> <artifactId>scalikejdbc-config_2.11</artifactId> <version>2.5.0</version> </dependency> <dependency> <groupId>com.typesafe</groupId> <artifactId>config</artifactId> <version>1.3.0</version> </dependency> <dependency> <groupId>org.apache.commons</groupId> <artifactId>commons-lang3</artifactId> <version>3.5</version> </dependency>
实现思路
1)StreamingContext 2)从kafka中获取数据(从外部存储获取offset-->根据offset获取kafka中的数据) 3)根据业务进行逻辑处理 4)将处理结果存到外部存储中--保存offset 5)启动程序,等待程序结束
代码实现
-
SparkStreaming主体代码如下
import kafka.common.TopicAndPartition import kafka.message.MessageAndMetadata import kafka.serializer.StringDecoder import org.apache.spark.SparkConf import org.apache.spark.streaming.kafka.{HasOffsetRanges, KafkaUtils} import org.apache.spark.streaming.{Seconds, StreamingContext} import scalikejdbc._ import scalikejdbc.config._ object JDBCOffsetApp { def main(args: Array[String]): Unit = { //创建SparkStreaming入口 val conf = new SparkConf().setMaster("local[2]").setAppName("JDBCOffsetApp") val ssc = new StreamingContext(conf,Seconds(5)) //kafka消费主题 val topics = ValueUtils.getStringValue("kafka.topics").split(",").toSet //kafka参数 //这里应用了自定义的ValueUtils工具类,来获取application.conf里的参数,方便后期修改 val kafkaParams = Map[String,String]( "metadata.broker.list"->ValueUtils.getStringValue("metadata.broker.list"), "auto.offset.reset"->ValueUtils.getStringValue("auto.offset.reset"), "group.id"->ValueUtils.getStringValue("group.id") ) //先使用scalikejdbc从MySQL数据库中读取offset信息 //+------------+------------------+------------+------------+-------------+ //| topic | groupid | partitions | fromoffset | untiloffset | //+------------+------------------+------------+------------+-------------+ //MySQL表结构如上,将“topic”,“partitions”,“untiloffset”列读取出来 //组成 fromOffsets: Map[TopicAndPartition, Long],后面createDirectStream用到 DBs.setup() val fromOffset = DB.readOnly( implicit session => { SQL("select * from hlw_offset").map(rs => { (TopicAndPartition(rs.string("topic"),rs.int("partitions")),rs.long("untiloffset")) }).list().apply() }).toMap //如果MySQL表中没有offset信息,就从0开始消费;如果有,就从已经存在的offset开始消费 val messages = if (fromOffset.isEmpty) { println("从头开始消费...") KafkaUtils.createDirectStream[String,String,StringDecoder,StringDecoder](ssc,kafkaParams,topics) } else { println("从已存在记录开始消费...") val messageHandler = (mm:MessageAndMetadata[String,String]) => (mm.key(),mm.message()) KafkaUtils.createDirectStream[String,String,StringDecoder,StringDecoder,(String,String)](ssc,kafkaParams,fromOffset,messageHandler) } messages.foreachRDD(rdd=>{ if(!rdd.isEmpty()){ //输出rdd的数据量 println("数据统计记录为:"+rdd.count()) //官方案例给出的获得rdd offset信息的方法,offsetRanges是由一系列offsetRange组成的数组 // trait HasOffsetRanges { // def offsetRanges: Array[OffsetRange] // } val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges offsetRanges.foreach(x => { //输出每次消费的主题,分区,开始偏移量和结束偏移量 println(s"---${x.topic},${x.partition},${x.fromOffset},${x.untilOffset}---") //将最新的偏移量信息保存到MySQL表中 DB.autoCommit( implicit session => { SQL("replace into hlw_offset(topic,groupid,partitions,fromoffset,untiloffset) values (?,?,?,?,?)") .bind(x.topic,ValueUtils.getStringValue("group.id"),x.partition,x.fromOffset,x.untilOffset) .update().apply() }) }) } }) ssc.start() ssc.awaitTermination() } }
-
自定义的ValueUtils工具类如下
import com.typesafe.config.ConfigFactory import org.apache.commons.lang3.StringUtils object ValueUtils { val load = ConfigFactory.load() def getStringValue(key:String, defaultValue:String="") = { val value = load.getString(key) if(StringUtils.isNotEmpty(value)) { value } else { defaultValue } } }
-
application.conf内容如下
metadata.broker.list = "192.168.137.251:9092" auto.offset.reset = "smallest" group.id = "hlw_offset_group" kafka.topics = "hlw_offset" serializer.class = "kafka.serializer.StringEncoder" request.required.acks = "1" # JDBC settings db.default.driver = "com.mysql.jdbc.Driver" db.default.url="jdbc:mysql://hadoop000:3306/test" db.default.user="root" db.default.password="123456"
-
自定义kafka producer
import java.util.{Date, Properties} import kafka.producer.{KeyedMessage, Producer, ProducerConfig} object KafkaProducer { def main(args: Array[String]): Unit = { val properties = new Properties() properties.put("serializer.class",ValueUtils.getStringValue("serializer.class")) properties.put("metadata.broker.list",ValueUtils.getStringValue("metadata.broker.list")) properties.put("request.required.acks",ValueUtils.getStringValue("request.required.acks")) val producerConfig = new ProducerConfig(properties) val producer = new Producer[String,String](producerConfig) val topic = ValueUtils.getStringValue("kafka.topics") //每次产生100条数据 var i = 0 for (i <- 1 to 100) { val runtimes = new Date().toString val messages = new KeyedMessage[String, String](topic,i+"","hlw: "+runtimes) producer.send(messages) } println("数据发送完毕...") } }
测试
-
启动kafka服务,并创建主题
[hadoop@hadoop000 bin]$ ./kafka-server-start.sh -daemon /home/hadoop/app/kafka_2.11-0.10.0.1/config/server.properties [hadoop@hadoop000 bin]$ ./kafka-topics.sh --list --zookeeper localhost:2181/kafka [hadoop@hadoop000 bin]$ ./kafka-topics.sh --create --zookeeper localhost:2181/kafka --replication-factor 1 --partitions 1 --topic hlw_offset
-
测试前查看MySQL中offset表,刚开始是个空表
mysql> select * from hlw_offset; Empty set (0.00 sec)
-
通过kafka producer产生500条数据
-
启动SparkStreaming程序
//控制台输出结果: 从头开始消费... 数据统计记录为:500 ---hlw_offset,0,0,500---
查看MySQL表,offset记录成功 mysql> select * from hlw_offset; +------------+------------------+------------+------------+-------------+ | topic | groupid | partitions | fromoffset | untiloffset | +------------+------------------+------------+------------+-------------+ | hlw_offset | hlw_offset_group | 0 | 0 | 500 | +------------+------------------+------------+------------+-------------+
-
关闭SparkStreaming程序,再使用kafka producer生产300条数据,再次启动spark程序(如果spark从500开始消费,说明成功读取了offset,做到了只读取一次语义)
//控制台结果输出: 从已存在记录开始消费... 数据统计记录为:300 ---hlw_offset,0,500,800---
-
查看更新后的offset MySQL数据
mysql> select * from hlw_offset; +------------+------------------+------------+------------+-------------+ | topic | groupid | partitions | fromoffset | untiloffset | +------------+------------------+------------+------------+-------------+ | hlw_offset | hlw_offset_group | 0 | 500 | 800 | +------------+------------------+------------+------------+-------------+
低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。
持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。
转载内容版权归作者及来源网站所有,本站原创内容转载请注明来源。
- 上一篇
多方面来保证Linux服务器的基本系统安全
作为一种开放源代码的操作系统,Linux服务器以其安全,高效和稳定的显著优势而得以广泛应用,但是,若不加以控制,也不见得安全到哪里,这篇博文主要从账号安全控制、系统引导和登录控制的角度,来进行Linux系统安全优化。并且使用辅助工具来查找安全隐患,以便我们运维人员及时采取相应的措施。 安全方面确实需要注意的比较多,这篇博文从各个方面写了下来,比较琐碎,还需要耐心些看,根据需要来配置服务器的安全性 一、基本安全措施: 1、 系统各种冗余账号,如“games”等,可直接删除,包括一些程序账号,若卸载程序后,账号没能被删除,则需要我们手动进行删除。 2、 当服务器中的用户账号已经固定,不再进行更改,可以直接锁定账号配置文件,锁定以后,便不可以添加用户及更改用户密码: [root@localhost ~]# chattr +i /etc/passwd /etc/shadow #锁定文件 [root@localhost ~]# lsattr /etc/passwd /etc/shadow #查看是否锁定 ----i----------- /etc/passwd ----i----------- ...
- 下一篇
服务器更换硬盘之后一定要检查是否同步
前言 近期再给一个用户更换服务器硬盘时发现,一是RAID配置错误,使用300GB的硬盘给600GB的磁盘做热备,另外是更换的硬盘型号不对导致数据没有同步。新近入职的管理员如果接手的有旧的服务器,有时间应重点检查一下硬盘及RAID信息,看是否有离线的硬盘或配置不合适的地方好及时纠正。 正文 今年4月上旬,有个单位的2台服务器硬盘坏了3块硬盘,客户让送3块600GB的硬盘并帮着给换一下。到了现场我看到了这两台服务器,其中一台服务器有2个硬盘亮黄灯(如图1所示),另一台服务器有1个硬盘亮黄灯(如图2所示)。 图1 2块硬盘亮黄灯 图2 1块硬盘亮黄灯 仔细一看,这两台服务器,每台都配置了1块300GB硬盘、3块600GB硬盘。当时比较奇怪怎么有这种配置,难道这块300GB的用来装系统、另3块600GB的做RAID-5存数据? 当时我和客户说,图1中这种坏了2块硬盘的可能需要做数据恢复了,因为3块硬盘的RAID-5中同时坏2块,此时换盘是不行了。当时这台服务器已经不能启动了(图2中的服务器还在使用,能进入系统)。 图1中坏2块硬盘的服务器,打开电源进入RAID界面,看到01、02槽位的硬盘为F...
相关文章
文章评论
共有0条评论来说两句吧...
文章二维码
点击排行
-
Docker使用Oracle官方镜像安装(12C,18C,19C)
- Springboot2将连接池hikari替换为druid,体验最强大的数据库连接池
- CentOS8编译安装MySQL8.0.19
- Docker快速安装Oracle11G,搭建oracle11g学习环境
- SpringBoot2配置默认Tomcat设置,开启更多高级功能
- MySQL8.0.19开启GTID主从同步CentOS8
- CentOS7,8上快速安装Gitea,搭建Git服务器
- Jdk安装(Linux,MacOS,Windows),包含三大操作系统的最全安装
- SpringBoot2编写第一个Controller,响应你的http请求并返回结果
推荐阅读
最新文章
- SpringBoot2编写第一个Controller,响应你的http请求并返回结果
- Docker使用Oracle官方镜像安装(12C,18C,19C)
- CentOS8编译安装MySQL8.0.19
- Springboot2将连接池hikari替换为druid,体验最强大的数据库连接池
- SpringBoot2整合Redis,开启缓存,提高访问速度
- CentOS关闭SELinux安全模块
- Hadoop3单机部署,实现最简伪集群
- CentOS6,7,8上安装Nginx,支持https2.0的开启
- SpringBoot2全家桶,快速入门学习开发网站教程
- SpringBoot2更换Tomcat为Jetty,小型站点的福音