是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结合的更多相关文章
-
Hadoop 2.7.4 HDFS+YRAN HA部署
实验环境 主机名称 IP地址 角色 统一安装目录 统一安装用户 sht-sgmhadoopnn-01 172.16.101.55 namenode,resourcemanager /usr/local ...
-
HA 高可用软件系统保养指南
又过了一年 618,六月是公司一年一度的大促月,一般提前一个月各系统就会减少需求和功能的开发,转而更多去关注系统可用性.稳定性和管控性等方面的非功能需求.大促前的准备工作一般叫作「备战」,可以把线上运 ...
-
MySQL: Fabric 搭建 HA
搭建好Fabric之后,就可以在它的基础上创建HA Group. Shard Group.HA+Shard Group等.这里来说明一下如何快速的搭建HA环境. Fabric 192.168.2.23 ...
-
zookeeper集群的搭建以及hadoop ha的相关配置
1.环境 centos7 hadoop2.6.5 zookeeper3.4.9 jdk1.8 master作为active主机,data1作为standby备用机,三台机器均作为数据节点,yarn资源 ...
-
使用Nginx+Lua代理Hadoop HA
一.Hadoop HA的Web页面访问 Hadoop开启HA后,会同时存在两个Master组件提供服务,其中正在使用的组件称为Active,另一个作为备份称为Standby,例如HDFS的NameNo ...
-
hadoop2.7.1 HA安装部署(转)
hadoop集群规划 目标:创建2个NameNode,做高可用,一个NameNode挂掉,另一个能够启动:一个运行Yarn,3台DataNode,3台Zookeeper集群,做高可用. 在 hadoo ...
-
【转】 XenServer架构之HA概述
一.XenServer HA概述 XenServer HA是一套全自动功能设计,规划,安全地恢复出现问题的XenServe 主机上的虚拟机的功能组件. 启用 HA 后,XenServer 将持续监视池 ...
-
ActiveMQ笔记(3):基于Networks of Brokers的HA方案
上一篇介绍了基于ZK的ActiveMQ HA方案,虽然理解起来比较容易,但是有二个不足: 1) 占用的节点数过多,1个zk集群至少3个节点,1个activemq集群也至少得3个节点,但其实正常运行时 ...
-
ActiveMQ笔记(2):基于ZooKeeper的HA方案
activemq官网给出了3种master/slave的HA方案,详见:http://activemq.apache.org/masterslave.html,基于共享文件目录,db,zookeepe ...
随机推荐
-
Python-基础数据类型
数据类型 计算机顾名思义就是可以做数学计算的机器,因此,计算机程序理所当然地可以处理各种数值.但是,计算机能处理的远不止数值,还可以处理文本.图形.音频.视频.网页等各种各样的数据,不同的数据,需要定 ...
-
NashZhou的自我介绍
行业: 电子商务服务业,目前主要是淘宝开放平台,ISV 关键词: 电商,淘宝直通车,关键词广告,自动优化 当前目标: 广告算法 广告主自动优化 希望能在这里结识有共同爱好踏实上进的园友,共同学习,共同 ...
-
Cheatsheet: 2013 09.01 ~ 09.09
.NET Multi Threaded WebScraping in CSharpDotNetTech .NET Asynchronous Patterns An Overview of Projec ...
-
UITableViewController 滚动引起的cocos2d动画暂停问题的解决
UITableViewController 滚动引起的cocos2d动画暂停问题的解决 之前在使用UITableViewController进行滚动时,cocos2d的动画会暂停,直至滚动完毕才会继续 ...
-
bootstrap-paginator 分页控件的使用
首先对js和css的引用 <link rel="stylesheet" href="reference/bootstrap/css/bootstrap.min.cs ...
-
C#基础:事件(二) 【转】
上篇文章介绍了C#中事件的基本实现方式,在本文中,将对最常见的事件委托EventHandler和EventHandler<T>做介绍. 事实上,在前面文章的介绍中,已经涉及到了EventH ...
-
制作流程图,activity,好不容易找到的
Star UML指导手册 Module by: Stephen Wong 原著:Stephen Wong 翻译:火猴 1. 综述:http://pan.baidu.com/s ...
-
Euclideanloss_layer层解析
这里说一下euclidean_loss_layer.cpp关于该欧式loss层的解析,代码如下: #include <vector> #include "caffe/layers ...
-
图解Fiddler如何抓取Android数据包
介绍Fiddler抓取Android数据包希望对大家的工作和学习有所帮助! 电脑开启wifi热点 首先在电脑上下载一个wifi软件,我这里用的是猎豹wifi,电脑开启wifi热点后,如下图所示: 设 ...
-
定期删除elasticsearch 的index 索引
#!/bin/bashfind /data/elasticsearch/data/pro-kz-log/nodes/0/indices/ -type d -mtime +7 | awk -F" ...