067 HA与updateStateByKey结合

时间:2022-09-06 15:00:56

  是HA与updateStateByKey相结合的程序。

  有点问题,有点奇怪,重启项目后运行没有问题,但是第三次启动的时候,就不会在打印数据了,有点问题。

1.程序

 package com.stream.it
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext} /**
* 通过HA管理KAFKA的offset偏移量
* Created by ibf on 03/04.
*/
object HAAndUpdateStateByKey {
def main(args: Array[String]): Unit = {
val conf = new SparkConf()
.setAppName("StreamingHAOfKafkaDirect")
.setMaster("local[*]")
val sc = SparkContext.getOrCreate(conf)
val checkpointDir = "hdfs://linux-hadoop01.ibeifeng.com:8020/beifeng/spark/streaming/012" // ========================================
/**
* 创建一个StreamingContext对象
*
* @return
*/
def createingStreamingContextFunc(): StreamingContext = {
val kafkaParams = Map("metadata.broker.list" -> "linux-hadoop01.ibeifeng.com:9092,linux-hadoop01.ibeifeng.com:9093,linux-hadoop01.ibeifeng.com:9094", "auto.offset.reset" -> "smallest")
val topics = Set("beifeng")
val ssc = new StreamingContext(sc, Seconds(1))
val dstream = KafkaUtils.createDirectStream[String, String, kafka.serializer.StringDecoder, kafka.serializer.StringDecoder](ssc, kafkaParams, topics) val resultDStream = dstream
.filter(_._2.nonEmpty)
.mapPartitions(iter => {
iter.flatMap(_._2.split(" ").map((_, 1)))
})
.updateStateByKey(
(values: Seq[Int], state: Option[Long]) => {
// 获取当前key的累加值
val currentSum = values.sum
// 获取之前批次的累加值
val preSum = state.getOrElse(0L) // 返回结果
Some(currentSum + preSum)
}
) // 打印一下
resultDStream.print() // ssc设置checkpoint
ssc.checkpoint(checkpointDir)
// 返回对象
ssc
} val ssc = StreamingContext.getOrCreate(
checkpointPath = checkpointDir,
creatingFunc = createingStreamingContextFunc
) // 将ssc传给其它管理Streaming关闭的对象==>其它线程(当前线程做数据的接收处理工作)
shutdownStreaming(ssc) // start ===spark.streaming.stopGracefullyOnShutdown为true的时候,当jvm退出的时候,自动关闭
ssc.start()
ssc.awaitTermination() // 阻塞操作
} /**
* 进行shutdown操作
*
* @param ssc
*/
def shutdownStreaming(ssc: StreamingContext): Unit = {
new Thread(new Runnable {
override def run(): Unit = {
println("你好......")
// 当某个条件被触发的时候,streamingcontext关闭
// 这里假设停留10秒中
Thread.sleep(10000)
// 关闭操作
ssc.stop(true, true)
println("你好")
}
}).start()
} }

2.效果

  067 HA与updateStateByKey结合

067 HA与updateStateByKey结合的更多相关文章

  1. Hadoop 2.7.4 HDFS+YRAN HA部署

    实验环境 主机名称 IP地址 角色 统一安装目录 统一安装用户 sht-sgmhadoopnn-01 172.16.101.55 namenode,resourcemanager /usr/local ...

  2. HA 高可用软件系统保养指南

    又过了一年 618,六月是公司一年一度的大促月,一般提前一个月各系统就会减少需求和功能的开发,转而更多去关注系统可用性.稳定性和管控性等方面的非功能需求.大促前的准备工作一般叫作「备战」,可以把线上运 ...

  3. MySQL: Fabric 搭建 HA

    搭建好Fabric之后,就可以在它的基础上创建HA Group. Shard Group.HA+Shard Group等.这里来说明一下如何快速的搭建HA环境. Fabric 192.168.2.23 ...

  4. zookeeper集群的搭建以及hadoop ha的相关配置

    1.环境 centos7 hadoop2.6.5 zookeeper3.4.9 jdk1.8 master作为active主机,data1作为standby备用机,三台机器均作为数据节点,yarn资源 ...

  5. 使用Nginx+Lua代理Hadoop HA

    一.Hadoop HA的Web页面访问 Hadoop开启HA后,会同时存在两个Master组件提供服务,其中正在使用的组件称为Active,另一个作为备份称为Standby,例如HDFS的NameNo ...

  6. hadoop2.7.1 HA安装部署(转)

    hadoop集群规划 目标:创建2个NameNode,做高可用,一个NameNode挂掉,另一个能够启动:一个运行Yarn,3台DataNode,3台Zookeeper集群,做高可用. 在 hadoo ...

  7. 【转】 XenServer架构之HA概述

    一.XenServer HA概述 XenServer HA是一套全自动功能设计,规划,安全地恢复出现问题的XenServe 主机上的虚拟机的功能组件. 启用 HA 后,XenServer 将持续监视池 ...

  8. ActiveMQ笔记(3):基于Networks of Brokers的HA方案

    上一篇介绍了基于ZK的ActiveMQ HA方案,虽然理解起来比较容易,但是有二个不足: 1)  占用的节点数过多,1个zk集群至少3个节点,1个activemq集群也至少得3个节点,但其实正常运行时 ...

  9. ActiveMQ笔记(2):基于ZooKeeper的HA方案

    activemq官网给出了3种master/slave的HA方案,详见:http://activemq.apache.org/masterslave.html,基于共享文件目录,db,zookeepe ...

随机推荐

  1. Python-基础数据类型

    数据类型 计算机顾名思义就是可以做数学计算的机器,因此,计算机程序理所当然地可以处理各种数值.但是,计算机能处理的远不止数值,还可以处理文本.图形.音频.视频.网页等各种各样的数据,不同的数据,需要定 ...

  2. NashZhou的自我介绍

    行业: 电子商务服务业,目前主要是淘宝开放平台,ISV 关键词: 电商,淘宝直通车,关键词广告,自动优化 当前目标: 广告算法 广告主自动优化 希望能在这里结识有共同爱好踏实上进的园友,共同学习,共同 ...

  3. Cheatsheet: 2013 09.01 ~ 09.09

    .NET Multi Threaded WebScraping in CSharpDotNetTech .NET Asynchronous Patterns An Overview of Projec ...

  4. UITableViewController 滚动引起的cocos2d动画暂停问题的解决

    UITableViewController 滚动引起的cocos2d动画暂停问题的解决 之前在使用UITableViewController进行滚动时,cocos2d的动画会暂停,直至滚动完毕才会继续 ...

  5. bootstrap-paginator 分页控件的使用

    首先对js和css的引用 <link rel="stylesheet" href="reference/bootstrap/css/bootstrap.min.cs ...

  6. C&num;基础:事件(二) 【转】

    上篇文章介绍了C#中事件的基本实现方式,在本文中,将对最常见的事件委托EventHandler和EventHandler<T>做介绍. 事实上,在前面文章的介绍中,已经涉及到了EventH ...

  7. 制作流程图,activity,好不容易找到的

    Star UML指导手册 Module by: Stephen Wong 原著:Stephen Wong            翻译:火猴 1.   综述:http://pan.baidu.com/s ...

  8. Euclideanloss&lowbar;layer层解析

    这里说一下euclidean_loss_layer.cpp关于该欧式loss层的解析,代码如下: #include <vector> #include "caffe/layers ...

  9. 图解Fiddler如何抓取Android数据包

    介绍Fiddler抓取Android数据包希望对大家的工作和学习有所帮助! 电脑开启wifi热点 首先在电脑上下载一个wifi软件,我这里用的是猎豹wifi,电脑开启wifi热点后,如下图所示:  设 ...

  10. 定期删除elasticsearch 的index 索引

    #!/bin/bashfind /data/elasticsearch/data/pro-kz-log/nodes/0/indices/ -type d -mtime +7 | awk -F&quot ...