【开发实例】spark 实战项目案例_spark wordcount实例_百度文库
1、本地运行实例1.package com.lin.wordcount2.import org.apache.spark.SparkConf精彩内容，尽在百度攻略：3.import org.apache.spark.SparkContext4.import org.apache.spark.SparkContext._5.精彩内容，尽在百度攻略：6./**7.* 本地运行Spark8.* 环境：Windows7精彩内容，尽在百度攻略：9.* scala版本：2.11.810.* Spark版本：1.6.111.*/精彩内容，尽在百度攻略：12.object WordCountLocal {13.14.def main(args: Array[String]): Unit = {精彩内容，尽在百度攻略：15./**16.* 第一步:创建Spark的配置对象SparkConf，设置Spark程序的运行时的配置信息17.* 例如说通过setMaster来设置程序要连接的Spark集群的Master的URL精彩内容，尽在百度攻略：18.* 如果设置为local，则代表Spark程序在本地运行，特别适合于配置条件的较差的人19.*20.*/精彩内容，尽在百度攻略：21.22.val conf = new SparkConf()23.conf.setAppName("wordCountLocal") //设置应用程序的名称，在程序运行的监控界面可以看到名称精彩内容，尽在百度攻略：24.conf.setMaster("local") //此时程序在本地运行，无需安装Spark的任何集群25.26./**精彩内容，尽在百度攻略：27.* 第二步:创建SparkContext对象28.* SparkContext是Spark程序所有功能的唯一入口，无论是采用Scala，Java，Python等都必须有一个SparkContext29.* SparkContext核心作用：初始化Spark应用程序运行所需要的核心组件，包括DAGScheduler，TaskScheduler，Scheduler精彩内容，尽在百度攻略：30.* 同时还会负责Spark程序往Master注册程序等31.* SparkContext是整个Spark应用程序中最为至关重要的一个对象。32.*/精彩内容，尽在百度攻略：33.34.val sc = new SparkContext(conf) //创建SparkContext对象，通过传入SparkConf实例来定制Spark运行的具体参数和配置信息35.精彩内容，尽在百度攻略：36./**37.* 第三步:根据具体的数据来源（HDFS，HBase，Local FS（本地文件系统） ，DB，S3（云上）等）通过SparkContext来创建RDD38.* RDD的创建基本有三种方式，根据外部的数据来源（例如HDFS），根据Scala集合，由其他的RDD操作产生精彩内容，尽在百度攻略：39.* 数据会被RDD划分成为一系列的Partitions，分配到每个Partition的数据属于一个Task的处理范畴40.*/41.精彩内容，尽在百度攻略：42.//文件的路径，最小并行度（根据机器数量来决定）43.//val lines:RDD[String]= sc.textFile("F://spark//spark-1.6.2-bin-hadoop2.6//README.md", 1)
//读取本地文件，并设置Partition = 144.val lines = sc.textFile("D://Java//spark//spark-1.6.1-bin-hadoop2.6//README.md", 1) //读取本地文件，并设置Partition = 1
//类型推导得出lines为RDD精彩内容，尽在百度攻略：45./**46.* 第四步:对初始的RDD进行Transformation级别的处理，例如map，filter等高阶函数等的编程，来进行具体的数据计算47.*
4.1:将每一行的字符串拆分成单个的单词精彩内容，尽在百度攻略：48.*
4.2:在单词拆分的基础上对每个单词的实例计数为1，也就是word =&(word,1)49.*
4.3:在每个单词实例计数为1基础之上统计每个单词在文件出现的总次数50.*/精彩内容，尽在百度攻略：51.52.//对每一行的字符串进行单词的拆分并把所有行的拆分结果通过flat合并成为一个大的单词集合53.val words = lines.flatMap { line =& line.split(" ") } //words同样是RDD类型精彩内容，尽在百度攻略：54.val pairs = words.map { word =& (word, 1) }55.val wordCounts = pairs.reduceByKey(_ + _) //对相同的key，进行value的累加（包括Local和Reducer级别同时Reduce）56.精彩内容，尽在百度攻略：57.wordCounts.foreach(wordNumberPair =& println(wordNumberPair._1 + " : " + wordNumberPair._2))58.59.sc.stop() //注意一定要将SparkContext的对象停止，因为SparkContext运行时会创建很多的对象精彩内容，尽在百度攻略：60.61./*这个程序运行之后一定会有一个错误，因为 没有hadoop环境，这个不是程序错误，也不影响任何功能*/62.精彩内容，尽在百度攻略：63.}64.}运行结果：精彩内容，尽在百度攻略：有可能会出现如下错误：精彩内容，尽在百度攻略：提示堆内存不够。右键-》run as -&run configurations 设置程序运行内存2、远程运行实例精彩内容，尽在百度攻略：1.package com.lin.wordcount2.3.import org.apache.spark.SparkConf精彩内容，尽在百度攻略：4.import org.apache.spark.SparkContext5.import org.apache.spark.SparkContext._6.精彩内容，尽在百度攻略：7./**8.* 在安装有Spark的Linux机器上跑9.* 环境：CentOS精彩内容，尽在百度攻略：10.* scala版本：2.11.811.* Spark版本：1.6.112.* hadoop版本：hadoop2.6.4精彩内容，尽在百度攻略：13.*/14.object WordCount {15.def main(args: Array[String]) {精彩内容，尽在百度攻略：16.if (args.length17.System.err.println("Usage: ")18.System.exit(1)精彩内容，尽在百度攻略：19.}20.21.val conf = new SparkConf()精彩内容，尽在百度攻略：22.val sc = new SparkContext(conf)23.val line = sc.textFile(args(0))24.精彩内容，尽在百度攻略：25.line.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_+_).collect().foreach(println)26.27.val result = line.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_+_)精彩内容，尽在百度攻略：28.result.collect().foreach(println)29.result.saveAsSequenceFile(args(1))30.精彩内容，尽在百度攻略：31.sc.stop()32.}33.}精彩内容，尽在百度攻略：看看pom文件：1.2.xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd"&精彩内容，尽在百度攻略：3.4.0.0<.lin5.Spark-Demo精彩内容，尽在百度攻略：6.0.0.1-SNAPSHOT7.${project.artifactId}8.My wonderfull scala app精彩内容，尽在百度攻略：9.201510.11.精彩内容，尽在百度攻略：12.My License13.http://....14.repo精彩内容，尽在百度攻略：15.16.17.精彩内容，尽在百度攻略：18.19.1.620.1.6精彩内容，尽在百度攻略：21.UTF-822.2.11.823.2.11精彩内容，尽在百度攻略：24.25.26.精彩内容，尽在百度攻略：27.28.org.scala-lang29.scala-library精彩内容，尽在百度攻略：30.${scala.version}31.32.精彩内容，尽在百度攻略：33.org.apache.spark34.spark-core_2.1135.1.6.1精彩内容，尽在百度攻略：36.37.38.精彩内容，尽在百度攻略：39.40.junit41.junit精彩内容，尽在百度攻略：42.4.1143.test44.精彩内容，尽在百度攻略：45.46.org.specs247.specs2-core_${pat.version}精彩内容，尽在百度攻略：48.2.4.1649.test50.精彩内容，尽在百度攻略：51.52.org.scalatest53.scalatest_${pat.version}精彩内容，尽在百度攻略：54.2.2.455.test56.精彩内容，尽在百度攻略：57.58.59.精彩内容，尽在百度攻略：60.src/main/scala61.src/test/scala62.精彩内容，尽在百度攻略：63.64.65.net.alchim31.maven精彩内容，尽在百度攻略：66.scala-maven-plugin67.3.2.068.精彩内容，尽在百度攻略：69.70.<pile精彩内容，尽在百度攻略：72.testCompile73.74.精彩内容，尽在百度攻略：75.76.-make:transitive --&77.-dependencyfile精彩内容，尽在百度攻略：78.${project.build.directory}/.scala_dependencies79.80.精彩内容，尽在百度攻略：81.82.83.精彩内容，尽在百度攻略：84.85.org.apache.maven.plugins86.maven-surefire-plugin精彩内容，尽在百度攻略：87.2.18.188.89.false精彩内容，尽在百度攻略：90.true91.92.**/*Test.*精彩内容，尽在百度攻略：93.**/*Suite.*94.95.精彩内容，尽在百度攻略：96.97.98.maven-assembly-plugin精彩内容，尽在百度攻略：99.2.6100.101.精彩内容，尽在百度攻略：102.jar-with-dependencies103.104.精彩内容，尽在百度攻略：105.106.107.精彩内容，尽在百度攻略：108.109.110.jdk-1.7精彩内容，尽在百度攻略：111.112.true113.1.7精彩内容，尽在百度攻略：114.115.116.1.7精彩内容，尽在百度攻略：117.1.7118.1.7119.精彩内容，尽在百度攻略：120.121.122.精彩内容，尽在百度攻略：123.项目目录结构：精彩内容，尽在百度攻略：运行;clean package打包：将整个jar包上传到linux目录精彩内容，尽在百度攻略：运行脚本：1../bin/spark-submit --master
spark://hmaster:7077 --class com.lin.wordcount.WordCount --executor-memory 1g /app/hadoop/spark-1.6.1/examples/Spark-Demo-0.0.1-SNAPSHOT.jar /input/LICENSE.txt /output精彩内容，尽在百度攻略：输出结果：同时将结果保存到hdfs中：精彩内容，尽在百度攻略：本文作者：evankaka君，已阅读到文档的结尾了呢~~
Spark中运行wordcount
扫扫二维码，随身浏览文档
手机或平板扫扫即可继续访问
Spark中运行wordcount
举报该文档为侵权文档。
举报该文档含有违规或不良信息。
反馈该文档无法正常浏览。
举报该文档为重复文档。
推荐理由：
将文档分享至：
分享完整地址
文档地址：
粘贴到BBS或博客
flash地址：
支持嵌入FLASH地址的网站使用
html代码：
&embed src='/DocinViewer--144.swf' width='100%' height='600' type=application/x-shockwave-flash ALLOWFULLSCREEN='true' ALLOWSCRIPTACCESS='always'&&/embed&
450px*300px480px*400px650px*490px
支持嵌入HTML代码的网站使用
您的内容已经提交成功
您所提交的内容需要审核后才能发布，请您等待！
3秒自动关闭窗口Spark安装并运行WordCount示例
Spark安装并运行WordCount示例。
1.官网下载spark包https://spark.apache.org/downloads.html
2.将spark安装包上传至集群，并解压。以下操作二选一。
3.部署standalone模式的spark集群。
修改conf/slaves文件，添加spark的各个节点ip地址。
4.Spark on yarn模式。
添加环境变量： HADOOP_CONF_DIR=hadoop配置文件所在路径。
2.下载与解压
tar zxvf spark-1.6.1-bin-hadoop2.6.tgz
mv spark-1.6.1-bin-hadoop2.6 spark-1.6.1
3.Local模式
中括号里面是分配的核数
读取文件，这里读取的是本地文件
scala代码如下：
val lines = sc.textFile(&/usr/local/data/a.txt&)
lines.flatMap(_.split(& &)).map((_, 1)).reduceByKey(_+_).collect.foreach(println)
进行wordcount并输出结果
查看spark UI
4.配置集群运行Standalone模式
1.复制slaves.template和spark-env.sh.template文件
cp slaves.template slaves
cp spark-env.sh.template spark-env.sh
2.配置slaves，把loalhost换成集群ip
3.配置spark-env.sh
4.scp 分发到集群
5.sbin目录下启动spark集群，成功后命令可以看到worker进程
6.进入bin目录下运行Standalone模式
7.读取文件（本次是从hdfs上读取）
cache()一下
查看读入数据的行数
查看UI，可以看到小绿点
8.进行wordcount操作
5.Yarn模式（yarn-client）
yarn-client：driver端跑在client端，可以看到日志与最终统计结果
Spark:Yarn-cluster和Yarn-client区别与联系请看这里
本次实验采用yarn-client
新建maven project
代码如下：
package com.neu.spark.sparkT
import java.util.ArrayL
import java.util.L
import java.util.M
import org.apache.spark.SparkC
import org.apache.spark.SparkC
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkC
import org.apache.spark.api.java.function.FlatMapF
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairF
import scala.Tuple2;
public class WordCount {
public static void main(String[] args) {
SparkConf conf = new SparkConf();
conf.setAppName(&wordcount&);
//conf.setMaster(&local[*]&);
SparkContext sc = new SparkContext(conf);
JavaSparkContext jsc = new JavaSparkContext(sc);
JavaRDD data =
jsc.textFile(&hdfs://172.17.11.172:9000/data/a.txt&);
JavaRDD wordRDD =data.flatMap(new FlatMapFunction() {
private static final long serialVersionUID = 8263126L;
public Iterable call(String t) throws Exception {
String[] words = t.split(& &);
List list = new ArrayList();
for (int i = 0; i & words. i++) {
list.add(words[i]);
JavaPairRDD
wordPairRDD= wordRDD.mapToPair(new PairFunction() {
private static final long serialVersionUID = 7010337L;
public Tuple2 call(String t) throws Exception {
// TODO Auto-generated method stub
return new Tuple2(t,1);
JavaPairRDD wordCountRS = wordPairRDD.reduceByKey(new Function2() {
public Integer call(Integer v1, Integer v2) throws Exception {
// TODO Auto-generated method stub
return v1.intValue()+v2.intValue();
Map result =wordCountRS.collectAsMap();
System.out.println(&The size is &+result.size());
for (Map.Entry entry:result.entrySet())
System.out.println(&Key = &+entry.getKey()+&, Value = &+entry.getValue());
配置文件如下：
com.neu.spark
0.0.1-SNAPSHOT
https://maven.apache.org
org.apache.spark
spark-core_2.10
org.apache.hadoop
hadoop-client
打成jar包，放在集群运行
这里注意参数，一开始报错，后来加上 class& 代码的包名.类名后成功了
读取的hdfs上的文件如下
运行结果如图：
查看8088端口
本次运行并没有设置参数，而是采用默认的，也可以设置参数，也可以集群运行，本地进行debug
/usr/local/tools/spark-1.6.1/bin/spark-submit \
--master yarn-client \
--driver-cores 8 \
--driver-memory 1G \
--num-executors 2 \
--executor-memory 1G \
--executor-cores 4 \
--driver-java-options '-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=9887'
/usr/local/userJars/spark/SparkTest-0.0.1.jar
#--driver-java-options '-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=9887' \
#-JAVA_OPTS = -Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=9888 \
6.问题总结
1.Error: No main class set in JAR; please specify one with &class
Run with &help for usage help or &verbose for debug output
加入class参数
2.org.apache.hadoop.mapred.InvalidInputException: Input path does not exist: file:/data/a.txt
但是写成sc.textFile(&hdfs:///data/a.txt&)或者sc.textFile(&file:///usr/local/data/a.txt&)仍然报错java.io.IOException: Incomplete HDFS URI, no host: hdfs:/data/a.txt
可能是hadoop配置文件没有写好，但是按照教程改好了之后还是报错，最终这种方式成功了
sc.textFile(&hdfs://172.17.11.172:9000/data/a.txt&);博客分类：
环境:
Hadoop版本：Apache Hadoop2.7.1
&&& Spark版本：Apache Spark1.4.1
核心代码：
测试数据：
a,b,a
在命令行使用sbt打包：sbt clean package
上传jar至Hadoop或者Spark的集群上，如何提交？
三种模式提交:
（1）需要启动HDFS+YRAN，无须启动spark的standalone集群
bin/spark-submit
--class com.spark.helloword.WordCount
--master yarn-client
./spark-hello_2.11-1.0.jar
（2）启动spark的standalone集群，并启动的Hadoop的HDFS分布式存储系统即可
bin/spark-submit
com.spark.helloword.WordCount
--master spark://h1:7077 ./spark-hello_2.11-1.0.jar
（3）//需要启动HDFS+YRAN，无须启动spark的standalone集群
//--name 指定作业名字
bin/spark-submit
--class com.spark.helloword.WordCount
--master yarn-cluster
--name test-spark-wordcount
./spark-hello_2.11-1.0.jar
执行结果：
(a,4)
(c,1)
运行模式截图：
最后欢迎大家扫码关注微信公众号：我是攻城师(woshigcs)，我们一起学习，进步和交流!（woshigcs）
本公众号的内容是有关搜索和大数据技术和互联网等方面内容的分享，也是一个温馨的技术互动交流的小家园，有什么问题随时都可以留言，欢迎大家来访！
qindongliang1922
浏览: 1053085 次
来自: 北京
浏览量：90443
浏览量：86469
浏览量：39329
浏览量：35686
有没有demo了，介绍了这么多 还是不知道怎么用和优势
内容非常不错，就是字体能修改一下就好了
楼主，请教个问题，文中提到的“索引两个字段，一个分词，一个不分 ...
向您请教，采用第二种方式启动Web项目，如何设置log，而不至 ...
非阻塞的举例不恰当后使用快捷导航没有帐号？
查看: 3495|回复: 4
请教Spark Streaming的例子NetworkWordCount在yarn上面运行时的问题
新手上路, 积分 21, 距离下一级还需 29 积分
论坛徽章:0
在spark官网上面关于spark streaming有这么一个例子，按照例子做也是没有问题的：
2.jpg (34.43 KB)
15:44 上传
之后我想把这个例子放到yarn上面去做，然后我用一下的命令：$ ./bin/spark-submit --class org.apache.spark.examples.streaming.NetworkWordCount --master yarn-cluster --num-executors 2 --driver-memory 512m --executor-memory 512m --executor-cores 1 lib/spark-examples*.jar localhost 9999
之后我在applications上面去看logs，一个stderr，一个stdout,然后按照例子来说，我在9999端口去输入数据的时候，应该在stdout会输出wordcount的结果，但是在stdout中永远只有一行，如下：
1.jpg (109.49 KB)
15:42 上传
不知道为什么，暂时自己也没找到解决办法，麻烦各位解答，我是spark的初学者，请各位帮忙，感激不尽！
[img]file:///C:UsersLuanPCAppDataRoamingTencentUsersQQWinTempRichOle9($3R[GQ1XETID(U$P1N22U.jpg[/img]
新手上路, 积分 21, 距离下一级还需 29 积分
论坛徽章:0
而且我在9999端口去输入数据的时候，在stderr中确实能看见数据过来了，如下：
3.jpg (50.72 KB)
15:49 上传
就是不知道为什么在stdout中不现实。求解！
新手上路, 积分 21, 距离下一级还需 29 积分
论坛徽章:0
没有高手能解决一下这个问题吗？顶！
新手上路, 积分 1, 距离下一级还需 49 积分
论坛徽章:0
我用run-example单机跑也没有结果，将run-example中的local[*]改为2以后可以，把--executor-cores 设为2试试吧
扫一扫加入本版微信群
