文档章节

Spark的JavaWordCount例子

q
 qinhui99
发布于 2017/03/04 14:24
字数 519
阅读 205
收藏 0

JavaWordCount的例子很简单,就是利用spark的API统计单词出现的次数。网络上很多人都拿它来当案例。但是,我没看到有人解释spark的API怎么做并发处理的细节。于是,我就自己动手做实验,增加了一些打印信息。果然,看到了多线程并发处理的情况。Spark程序不是串行地处理,而是在收集了两行数据之后就开始多线程并发处理。有些和CUDA程序类似。

JAVA的主要代码:

//处理每一行数据的函数

JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {

      @Override

      public Iterator<String> call(String s) {

      System.out.println("FlatMapFunction s="+s);

        return Arrays.asList(SPACE.split(s)).iterator();

      }

    });

//分割每一行成多个key

    JavaPairRDD<String, Integer> ones = words.mapToPair(

      new PairFunction<String, String, Integer>() {

        @Override

        public Tuple2<String, Integer> call(String s) {

        System.out.println("PairFunction s="+s);

          return new Tuple2<>(s, 1);

        }

      });

//根据key归约统计数字

    JavaPairRDD<String, Integer> counts = ones.reduceByKey(

      new Function2<Integer, Integer, Integer>() {

        @Override

        public Integer call(Integer i1, Integer i2) {

        System.out.println("reduceByKey count="+(i1+i2));

          return i1 + i2;

        }

      });

整段代码看起来像是串行执行的,实际上是并发执行的。

 

需要统计的文件内容:

good bad cool

hadoop spark mlib good

good spark mlib spark

cool spark bad

 

JavaWordCount的运行输出:

FlatMapFunction s=good bad cool

PairFunction s=good

PairFunction s=bad

PairFunction s=cool

FlatMapFunction s=hadoop spark mlib good

PairFunction s=hadoop

PairFunction s=spark

PairFunction s=mlib

PairFunction s=good

reduceByKey count=2

FlatMapFunction s=good spark mlib spark

PairFunction s=good

reduceByKey count=3

PairFunction s=spark

reduceByKey count=2

PairFunction s=mlib

reduceByKey count=2

PairFunction s=spark

reduceByKey count=3

FlatMapFunction s=cool spark bad

PairFunction s=cool

reduceByKey count=2

PairFunction s=spark

reduceByKey count=4

PairFunction s=bad

reduceByKey count=2

 

从上面的打印信息可以看到,当FlatMapFunction处理完第二行数据的时候,reduceByKey线程就开始工作了。直接进入并发模式,不需要等到FlatMapFunction处理完所有的行。这样计算的效率大幅提高,节省了很多等待时间。

值得注意的是,hadoop这个单词只出现了一次,结果它没有触发reduceByKey的call函数调用。说明这个call函数是针对两个输入数据进行归约处理的。

 

© 著作权归作者所有

q
粉丝 66
博文 73
码字总数 34091
作品 0
深圳
程序员
私信 提问
Spark源码分析调试环境搭建

目前常用的Spark版本有三种Cloudera、HDP和Apache,源码的获取方式可以在各自官网下载。本文选择Apache版本。 搭建环境所需要的工具如下: CentOS 7 maven 3.5.0 Java 1.8.0 Scala 2.12.2 I...

火力全開
2017/10/26
37
0
Spark安装启动 and 在程序中调用spark服务

1.软件准备 我的系统环境为ubuntu 13.10 1.scala-2.9.3.tgz http://www.scala-lang.org/files/archive/scala-2.9.3.tgz 由于spark是依赖scala2.9.3开发的,所以只能安装2.9.3,我试的时候一开...

zachary124
2014/01/25
18.4K
9
好程序员大数据教程:SparkShell和IDEA中编写Spark程序

好程序员大数据教程:SparkShell和IDEA中编写Spark程序,spark-shell是Spark自带的交互式Shell程序,方便用户进行交互式编程,用户可以在该命令行下用Scala编写Spark程序。spark-shell程序一...

好程序员IT
05/28
38
0
附录A Spark2.1核心工具类Utils

注:本文是为了配合《Spark内核设计的艺术 架构设计与实现》一书的内容而编写,目的是为了节省成本、方便读者查阅。书中附录A的内容都在本文呈现。 Utils是Spark最常用的工具类之一,Spark C...

beliefer
2018/01/25
0
0
Kubernetes助力Spark大数据分析

Kubernetes 作为一个广受欢迎的开源容器协调系统,是Google于2014年酝酿的项目。从Google趋势上看到,Kubernetes自2014年以来热度一路飙升,短短几年时间就已超越了大数据分析领域的长老Had...

店家小二
2018/12/17
0
0

没有更多内容

加载失败,请刷新页面

加载更多

Taro 兼容 h5 踩坑指南

最近一周在做 Taro 适配 h5 端,过程中改改补补,好不酸爽。 本文记录📝遇到的问题,希望为有相同需求的哥们👬节约点时间。 Taro 版本:1.3.9。 解决跨域问题 h5 发请求会报跨域问题,需...

dkvirus
今天
4
0
Spring boot 静态资源访问

0. 两个配置 spring.mvc.static-path-patternspring.resources.static-locations 1. application中需要先行的两个配置项 1.1 spring.mvc.static-path-pattern 这个配置项是告诉springboo......

moon888
今天
3
0
hash slot(虚拟桶)

在分布式集群中,如何保证相同请求落到相同的机器上,并且后面的集群机器可以尽可能的均分请求,并且当扩容或down机的情况下能对原有集群影响最小。 round robin算法:是把数据mod后直接映射...

李朝强
今天
4
0
Kafka 原理和实战

本文首发于 vivo互联网技术 微信公众号 https://mp.weixin.qq.com/s/bV8AhqAjQp4a_iXRfobkCQ 作者简介:郑志彬,毕业于华南理工大学计算机科学与技术(双语班)。先后从事过电子商务、开放平...

vivo互联网技术
今天
19
0
java数据类型

基本类型: 整型:Byte,short,int,long 浮点型:float,double 字符型:char 布尔型:boolean 引用类型: 类类型: 接口类型: 数组类型: Byte 1字节 八位 -128 -------- 127 short 2字节...

audience_1
今天
9
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部