文档章节

Spark Streaming(5):Spark Window function in Java

Joe_Wu
 Joe_Wu
发布于 2017/08/10 17:58
字数 287
阅读 9
收藏 0

首先,看下window函数的图解:

下面这个代码是计算一分钟之内的单词数量统计,每两秒获取一次数据,同时处理数据时间也是两秒,窗口大小为1分钟

1.数据源

package com.ssm.test;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.ServerSocket;
import java.net.Socket;

public class SocketTest {

	public static void main(String[] args) {
		try{
            ServerSocket server=new ServerSocket(9999);
            System.out.println("The Server has started");
            
            Socket socket=server.accept();
            System.out.println("There was a socket client connected");
            
            BufferedReader br=new BufferedReader(new InputStreamReader(System.in));
            String line=br.readLine();
            PrintWriter writer=new PrintWriter(socket.getOutputStream());
            
            while(!line.equals("end")){
            	writer = new PrintWriter(socket.getOutputStream());
    			writer.println("If you are not brave enough, no one will back you up.");
                writer.flush();
                System.out.println(System.currentTimeMillis() + "\nServer send:\t"+line);
                Thread.sleep(2000);
            }
            writer.close();
            socket.close();
            server.close();
        }catch(Exception e) {
        	e.printStackTrace();
        }
	}
}

2.spark streaming处理代码

package com.ssm.test;

import java.util.Arrays;
import java.util.Iterator;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.FlatMapFunction;
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.JavaDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;

import scala.Tuple2;

public class WordsCountTest {

	public static void main(String[] args) throws Exception {
		
		SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("WordsCount");
		JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(2));
		JavaDStream<String> lines = jssc.socketTextStream("127.0.0.1", 9999).window(new Duration(60000));
		
		lines.flatMap(new FlatMapFunction<String, String>(){
			public Iterator<String> call(String x){
				return Arrays.asList(x.split(" ")).iterator();
			}
		}).mapToPair(new PairFunction<String, String, Integer>(){
			public Tuple2<String, Integer> call(String s){
				return new Tuple2<String, Integer>(s, 1);
			}
		}).reduceByKey(new Function2<Integer, Integer, Integer>(){
			public Integer call(Integer i1, Integer i2){
				return i1+i2;
			}
		}).print();
		
		jssc.start();
		jssc.awaitTermination();

	}

}

 

© 著作权归作者所有

共有 人打赏支持
Joe_Wu
粉丝 1
博文 9
码字总数 6567
作品 0
徐汇
程序员
私信 提问
Spark2.1.0之基础知识

版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/beliefer/article/details/80303035 在阅读本文之前,读者最好已经阅读了《Spark2.1.0之初识Spark》一文,本文...

泰山不老生
05/24
0
0
Spark 2.0 时代全面到来 —— 2.0.1 版本发布

距离Spark 2.0.0发布两个月后,Spark 2.0.1版本发布了,这是一个修正版本,共处理了300多个Issue,涉及spark稳定性和bug等方面的修复 ,它的发布意味着Spark 2.0接近生产环境使用要求,想要尝...

达尔文
2016/10/08
13.1K
22
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
Apache Spark 2.4.0 正式发布

Apache Spark 2.4 与昨天正式发布,Apache Spark 2.4 版本是 2.x 系列的第五个版本。 如果想及时了解 Spark、Hadoop或者Hbase相关的文章,欢迎关注微信公共帐号: itebloghadoop Apache Spa...

Spark
11/09
0
0
Spark 学习资源收集【Updating】

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

大数据之路
2014/09/08
0
1

没有更多内容

加载失败,请刷新页面

加载更多

zookeeper配置与使用

一.登录官网下载 不要带后缀的,那是公侧版本,下稳定版,比如3.4.9 二.安装与使用 解压后bin里是启动程序 配置文件:在conf下 复制zoo_sample.cfg改名为为zoo.cfg,打开zoo修改文件...

小兵胖胖
27分钟前
2
0
spring源码阅读笔记(一)

ClassPathXmlApplicationContext 与 FileSystemXmlApplicationContext 用了这么久的框架,是时候搞一下源码了,一般最初接触spring 从以下步骤开始 创建一个bean类 并创建 ooxx.xml之类的spr...

NotFound403
50分钟前
4
0
MySQL主从配置

12月14日任务 17.1 MySQL主从介绍 17.2 准备工作 17.3 配置主 17.4 配置从 17.5 测试主从同步 MySQL主从介绍 MySQL主从又叫做Replication、AB复制。简单将就是A/B两个服务器做主从后,在A上写...

robertt15
53分钟前
10
0
我的Linux系统九阴真经

在今天,互联网的迅猛发展,科技技术也日新月异,各种编程技术也如雨后春笋一样,冒出尖来了。各种创业公司也百花齐放百家争鸣,特别是针对服务行业,新型互联网服务行业,共享经济等概念的公...

问题终结者
今天
22
0
Java 使用 gson 对 json 根据 key 键进行排序

引入Google的gson jar <dependency> <groupId>com.google.code.gson</groupId> <artifactId>gson</artifactId> <version>2.8.0</version>......

yh32
今天
4
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部