文档章节

Spark Streaming(4):Spark updateStateByKey in Java

Joe_Wu
 Joe_Wu
发布于 2017/08/10 12:29
字数 349
阅读 16
收藏 0
package com.pyrrha.examples;

import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;

import kafka.serializer.StringDecoder;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.Optional;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaPairInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;

import scala.Tuple2;

public class updateStateByKey {
	
	private static final String KAFKA_TOPIC = "TopicA";
	
	public static void main(String[] args) throws Exception {
		System.setProperty("hadoop.home.dir", "D:\\checkpoint\\hadoop-common-2.2.0-bin-master");
		SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("WordsCount");
		JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.milliseconds(2000));

		Map<String, String> kafkaParams = new HashMap<String, String>();
		kafkaParams.put("bootstrap.servers", "127.0.0.1:9092");
		kafkaParams.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		kafkaParams.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		kafkaParams.put("group.id", "lingroup");
		//kafkaParams.put("auto.offset.reset", "latest");
		
		Set<String> topics = new HashSet<String>();
        topics.add(KAFKA_TOPIC);
		
        JavaPairInputDStream<String, String> stream = org.apache.spark.streaming.kafka.KafkaUtils.
			createDirectStream(jssc, String.class, String.class, StringDecoder.class, StringDecoder.class, kafkaParams, topics);
		
        JavaPairDStream<String, Integer> transDStream = stream.flatMap(new FlatMapFunction<Tuple2<String,String>, String>() {
			public Iterator<String> call(Tuple2<String, String> t) throws Exception {
				return Arrays.asList(t._2.split(" ")).iterator();
			}
		}).filter(new Function<String, Boolean>() {
			public Boolean call(String v1) throws Exception {
				return v1.equals("a") ? false :true;
			}
		}).mapToPair(new PairFunction<String, String, Integer>() {
			public Tuple2<String, Integer> call(String t) throws Exception {
				return new Tuple2<String, Integer>(t, 1);
			}
		}).reduceByKey(new Function2<Integer, Integer, Integer>() {
			public Integer call(Integer v1, Integer v2) throws Exception {
				return v1 + v2;
			}
		}).window(new Duration(6000), new Duration(6000));
		
        /**
         * @param v1表示从上面获取的transDStream的所有value的List集合(如果transDStream=["b":2,"c":1],那么v1=[2,1])
         * @param v2表示上一个updateStateByKey的每个value保存的值
         * 所以,v1就是每次最新batch处理后的value集合,v2就是上个batch处理后缓存的value值
         */
        transDStream.updateStateByKey(new Function2<List<Integer>, Optional<Integer>, Optional<Integer>>() {
			public Optional<Integer> call(List<Integer> v1, Optional<Integer> v2) throws Exception {
				Integer v3 = 0;
				if(v2.isPresent())
					v3 = v2.get();
				for (Integer v : v1)
					v3 += v;
				
				return Optional.of(v3);
			}
		}).print();
        
		jssc.checkpoint("file:///D:/checkpoint/");
		jssc.start();
		jssc.awaitTermination();
	}
	
	

}

 

© 著作权归作者所有

共有 人打赏支持
Joe_Wu
粉丝 1
博文 9
码字总数 6567
作品 0
徐汇
程序员
Spark 学习资源收集【Updating】

(一)spark 相关安装部署、开发环境 1、Spark 伪分布式 & 全分布式 安装指南 http://my.oschina.net/leejun2005/blog/394928 2、Apache Spark探秘:三种分布式部署方式比较 http://dongxic...

大数据之路
2014/09/08
0
1
Spark2.1.0之基础知识

在阅读本文之前,读者最好已经阅读了《Spark2.1.0之初识Spark》一文,本文将对Spark的基础知识进行介绍。但在此之前,读者先跟随本人来一次简单的时光穿梭,最后还将对Java与Scala在语言上进...

beliefer
05/24
0
0
Spark Streaming 基本概念

介绍 Spark Streaming架构图 the micro-batch architecture of Spark Streaming Execution of Spark Streaming within Spark’s components JAVA代码示例 执行方式 1:修改log4j的日志级别为......

cloud-coder
2015/06/18
0
0
3.sparkSQL整合Hive

  spark SQL经常需要访问Hive metastore,Spark SQL可以通过Hive metastore获取Hive表的元数据。从Spark 1.4.0开始,Spark SQL只需简单的配置,就支持各版本Hive metastore的访问。注意,涉...

intsmaze(刘洋)
08/09
0
0
spark出现GC overhead limit exceeded和java heap space

spark执行任务时出现java.lang.OutOfMemoryError: GC overhead limit exceeded和java.lang.OutOfMemoryError: java heap space 最直接的解决方式就是在spark-env.sh中将下面两个参数调节的尽...

闵开慧
2014/10/14
0
1

没有更多内容

加载失败,请刷新页面

加载更多

5、前后端分离跨域问题

在以往的开发中,前后端分离也不是像现在这么热门,所谓的前端工程师也只不过是写好静态页面由Java工程师或者php工程师嵌入到页面中进行开发,这或许加重了这些工程师的工作量,而且在样式调...

永远的Chester
12分钟前
0
0
全志T3 Linux显示驱动分析

1、总体架构 全志T3处理器的显示框架是基于标准Linux的帧缓冲架构,其结构如图 1.1所示。显示控制器DE的驱动架构如图 1.2所示,包括屏蔽差异的显示管理抽象层,以及显示图层驱动、显示设备驱...

pnsam301
20分钟前
0
0
【HAVENT原创】VUE2 经验问题汇总

新建一个 Vue 实例可以有下列两种方式: 1. new 一个实例 var app= new Vue({ el:'#todo-app', // 挂载元素 data:{ // 在.vue组件中data是一个函数,要写成data () {}这种方式 i...

HAVENT
23分钟前
0
0
IO基础知识

读文件,可以使用内置的open(file,mode); with 语句自动close; 查看当前目录的绝对路径:os.path.abspath('.'); 创建目录:os.mkdir(path); 删除目录:os.rmdir(path); 拼接路径推荐使用:o...

年轻的中年大叔
26分钟前
0
0
BATJ等大厂最全经典面试题分享

金九银十,又到了面试求职高峰期,最近有很多网友都在求大厂面试题。正好我之前电脑里面有这方面的整理,于是就发上来分享给大家。 这些题目是网友去百度、蚂蚁金服、小米、乐视、美团、58、...

老道士
28分钟前
0
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部