在sparkStreaming实时存储时的问题

Posted 2021-02-13 hejunhong

1.实时插入mysql时遇到的问题，使用的updateStaeBykey有状态的算子 必须设置checkpoint  如果报错直接删掉checkpoint

在创建的时候自己保存偏移量即可 再次启动时读取正确偏移量就行了 管他checkpoint  无关的事了

实时插入时有个问题是怎么进行mysql的数据覆盖 掉一批次的值：

1.使用局部更新的sql ：

insert into area_user_amt (date,country,provence,amt)  values(‘${datekey}‘,‘${countrykey}‘,‘${provencekey}‘,‘${amt}‘) ON DUPLICATE KEY UPDATE `amt`= ‘${amt}‘

2.使用replace 相当于先删除在插入

replace into stream_offset(topic,partitions,groupid,brokerlist,offset)values (?,?,?,?,?)

2.使用redis 不使用叠加状态的updateStaeBykey ，进行完reduceBykey(list1，list2)=>(list.zip(list2)).map(_.1+_.2)  reduceBykey的两个参（累计值，当前值）一直做zip操作，做完后
（10,1）.zip（20,2）=》（(10,20),(1,2)）在做map对里面每一个进行相加就是累加值 （只是当前批次的）

使用redis的hincrby 值增加的方法实现 累加求和

.foreachPartition(iter=>{  
 //在各分区获取redis连接   
val jedis=JedisUtil.getJedisClient()   
iter.foreach(tp=>{     
//B2019040114 ,成功量 ,总量    
 jedis.hincrBy("P-"+tp._1._1.substring(0,8),tp._1._2,tp._2(0).toLong)
    //设置key的有效时间    
 jedis.expire(tp._1._1,60*60*24*7)   })   jedis.close()
 })

 

 

 

 

SparkStreaming使用checkpoint存在的问题

SparkStreaming在处理kafka中的数据时，存在一个kafka offset的管理问题：

• 官方的解决方案是checkpoint:
  • checkpoint是对sparkstreaming运行过程中的元数据和 每次rdds的数据状态保存到一个持久化系统中，当然这里面也包含了offset，一般是HDFS,S3，如果程序挂了，或者集群挂了，下次启动仍然能够从checkpoint中恢复，从而做到生产环境的7*24高可用。如果checkpoint存储做hdfs中，会带来小文件的问题。

但是checkpoint的最大的弊端在于，一旦你的流式程序代码或配置改变了，或者更新迭代新功能了，这个时候，你先停旧的sparkstreaming程序，然后新的程序打包编译后执行运行，会出现两种情况：

• （1）启动报错，反序列化异常
• （2）启动正常，但是运行的代码仍然是上一次的程序的代码。

为什么会出现上面的两种情况？

• 这是因为checkpoint第一次持久化的时候会把整个相关的jar给序列化成一个二进制文件，每次重启都会从里面恢复，但是当你新的 程序打包之后序列化加载的仍然是旧的序列化文件，这就会导致报错或者依旧执行旧代码。有的同学可能会说，既然如此，直接把上次的checkpoint删除了，不就能启动了吗？ 确实是能启动，但是一旦你删除了旧的checkpoint，新启动的程序，只能从kafka的smallest或者largest的偏移量消费，默认是从最新的，如果是最新的，而不是上一次程序停止的那个偏移量 就会导致有数据丢失，如果是老的，那么就会导致数据重复。不管怎么样搞，都有问题。