3932人阅读
大数据（23）
这篇教程为使用spark提供一个快速的介绍。我们将先介绍spark shell的API(python or scala),然后展示如何用JAVA,PYTHON,SCALA写应用。
请先安装SPARK，下载地址http://spark.apache.org/downloads.html，由于我们不使用HDFS，可以基于任何hadoop版本。
通过spark shell进行交互式分析
启动spark shell
./bin/spark-shell
可以看到，交互shell已经为我们准备好了Spark context,变量名为sc,现在用它来提交一个作业。
RDD是Spark的一个关键概念，它对分布式数据集合进行了抽象，称为弹性分布式数据。RDDs可以由Hadoop的输入数据，比如HDFS文件创建，也可以由其它RDDs转换而来。下面由README.md文本文件创建了一个RDD.
RDD有动作，返回一个值。也有转换，返回一个新的RDDs.我们先来执行一些动作。
我们再来执行一次转换，我们使用filter转换操.返回一个新RDD，它包含了文本文件的一个子集合,只包含&Spark&字符串的行。
我们也可以连续进行转换和执行动作
更多的RDD操作
RDD的动作和转换可以进行更复杂的计算。我们现在找出包含单词最多的行。
第一个map操作将每一行映射为这一行的单词数，创建了一个新RDD。然后reduce操作找出单词数最多的行。map,reduce的参数是一个scala函数（闭包）,我们可以使用任何Scala/Java的语言特性。比如，我们可以很容易的调用声明的函数，我们将使用Math.max函数将代码变得更易读。
另外一个常用的数据计算模型MapReduce,基于RDD模型，Spark也可以很容易的实现.
这里，我们组合使用flatMap,map和reduceByKey转换来计算文件中每个单词的个数。文件被抽象为一个(String,Int)元组的RDD。我们用collect操作在spark shell里收集每个单词的个数
Spark也支持将数据提取到集群范围内的缓存中。这对频繁访问的数据十分有用。比如，我们访问一小组热点的数据在PageRank这种迭代算法中。作为一个简单的示例，让我们把我们的lineWithSpark数据集合标记为缓存的。
将一个几百行的文本文件用spark来浏览和缓存看起来很愚蠢。但它是看起来很有趣，因为同样的函数可以用在每一个大的数据集上，而不用关心它们分布的数百上行的节点之上。
自己写应用程序
我们现在用spark API写一个应用程序.这个简单的应用程序,可以使用SBT(SCALA),Maven(JAVA),Python.
先用SCALA写一个示例
这个程序只是简单的统计文件中包含'a'和'b'的行数。YOUR_SPARK_HOME是你的spark安装目录。
我们的程序依懒spark api的lib所以还要编辑simple.sbt配置文件:
注意，我们使用sbt构建这个程序，所以要先下载安装sbt.并且程序的目录结构要符合sbt的约定。
最后我们用sbt打包刚才的程序，并用spark-submit提交运行
http://spark.apache.org/docs/latest/quick-start.html
&&相关文章推荐
参考知识库
* 以上用户言论只代表其个人观点，不代表CSDN网站的观点或立场
访问：338852次
积分：5051
积分：5051
排名：第5057名
原创：173篇
转载：37篇
评论：45条
(7)(8)(11)(14)(21)(15)(6)(1)(3)(4)(11)(3)(2)(3)(7)(3)(7)(12)(8)(5)(1)(10)(1)(1)(8)(1)(2)(4)(5)(5)(27)(2)1、本地运行成功，但是上传到服务器上产生了如下错误：
Exception in thread &main& java.lang.NoSuchMethodError: scala.runtime. ObjectRef.create (Ljava/lang/O) Lcala/runtime/ObjectR
此类问题往往是由于本地编译scala程序的scala版本和服务器上的版本不统一导致，可以通过更改本地的scala版本号来进行编译。
一般可以通过在服务上执行：more &/etc/profile 查看scala版本，Hadoop版本，spark版本。尽量保持jdk版本也能够统一(没有对jdk版本不一致引发的错误进行具体分析)
2、另外由于证书问题报错：
submit 一个jar文件的时候，会遇到以下报错
报错类型：
Invalid signature file digest for Manifest main attributes
错误产生原因：
jar文件中的META-INF目录中包含一个错误证书，在Linux服务器中执行以下命令：
zip -d &jar file name&.jar META-INF/*.RSA META-INF/*.DSA META-INF/*.SF
3. 曾经犯过一个白痴的问题，在Windows下编辑的shell脚本，直接rz 到服务器上，出现了各种bug,&\&等参数都不支持。
Unix系统中，每行结尾只有“&换行&”，即“\n”,Windows中，每行结尾是&\r\n&。Mac系统中，每行结尾是“&\r&”。Unix/Max系统下的文件在windows中打开的话，所有文字变成一行。
Windows中的文件在Unix/Mac下打开的话，每行的结尾会多出一个^M符号。
如果非要将windows下的文件转化为Unix下的文件的话，在Linux系统中，执行yum install dos2unix。dos2unix命令会将该文件转化为Unix环境识别的文件。
4、如何给java/scala程序的main传入参数
命令行中运行：a、首先需要编译程序：javac *.java； b、java * args1,args2 .....在ide中，run-&Edit Configuration-&Program arguments
&&相关文章推荐
参考知识库
* 以上用户言论只代表其个人观点，不代表CSDN网站的观点或立场
访问：185次
排名：千里之外中国领先的IT技术网站
51CTO旗下网站
利用Scala语言开发Spark应用程序
park内核是由Scala语言开发的，因此使用Scala语言开发Spark应用程序是自然而然的事情。如果你对Scala语言还不太熟悉，可以阅读网络教程A Scala Tutorial for Java Programmers或者相关Scala书籍进行学习。
作者：佚名来源：dongxicheng.org| 10:12
Spark内核是由Scala语言开发的，因此使用Scala语言开发Spark应用程序是自然而然的事情。如果你对Scala语言还不太熟悉，可以阅读网络教程A Scala Tutorial for Java Programmers或者相关Scala书籍进行学习。
本文将介绍3个Scala Spark编程实例，分别是WordCount、TopK和SparkJoin，分别代表了Spark的三种典型应用。
1. WordCount编程实例
WordCount是一个最简单的分布式应用实例，主要功能是统计输入目录中所有单词出现的总次数，编写步骤如下：
步骤1：创建一个SparkContext对象，该对象有四个参数：Spark
master位置、应用程序名称，Spark安装目录和jar存放位置，对于Spark On
YARN而言，最重要的是前两个参数，第一个参数指定为&yarn-standalone&，第二个参数是自定义的字符串，举例如下：
val&sc&=&new&SparkContext(args(0),&&WordCount&,&&&&&System.getenv(&SPARK_HOME&),&Seq(System.getenv(&SPARK_TEST_JAR&)))&&
步骤2：读取输入数据。我们要从HDFS上读取文本数据，可以使用SparkCon
val&textFile&=&sc.textFile(args(1))&
当然，Spark允许你采用任何Hadoop InputFormat，比如二进制输入格式SequenceFileInputFormat，此时你可以使用SparkContext中的hadoopRDD函数，举例如下：
val&inputFormatClass&=&classOf[SequenceFileInputFormat[Text,Text]]&var&hadoopRdd&=&sc.hadoopRDD(conf,&inputFormatClass,&classOf[Text],&classOf[Text])&
或者直接创建一个HadoopRDD对象：
var&hadoopRdd&=&new&HadoopRDD(sc,&conf,&&&&&&classOf[SequenceFileInputFormat[Text,Text,&classOf[Text],&classOf[Text])&
步骤3：通过RDD转换算子操作和转换RDD，对于WordCount而言，首先需要从输入数据中每行字符串中解析出单词，然后将相同单词放到一个桶中，最后统计每个桶中每个单词出现的频率，举例如下：
val&result&=&hadoopRdd.flatMap{&&&&&&&&&case(key,&value)&&=&value.toString().split(&\\s+&);&}.map(word&=&(word,&1)).&reduceByKey&(_&+&_)&
其中，flatMap函数可以将一条记录转换成多条记录（一对多关系），map函数将一条记录转换为另一条记录（一对一关系），reduceByKey函数将key相同的数据划分到一个桶中，并以key为单位分组进行计算，这些函数的具体含义可参考：Spark Transformation。
步骤4：将产生的RDD数据集保存到HDFS上。可以使用SparkContext中的saveAsTextFile哈数将数据集保存到HDFS目 录下，默认采用Hadoop提供的TextOutputFormat，每条记录以&（key,value）&的形式打印输出，你也可以采用 saveAsSequenceFile函数将数据保存为SequenceFile格式等，举例如下：
result.saveAsSequenceFile(args(2))&
当然，一般我们写Spark程序时，需要包含以下两个头文件：
import&org.apache.spark._&import&SparkContext._&
WordCount完整程序已在&Apache Spark学习：利用Eclipse构建Spark集成开发环境&一文中进行了介绍，在次不赘述。
需要注意的是，指定输入输出文件时，需要指定hdfs的URI，比如输入目录是hdfs://hadoop-test/tmp/input，输出目 录是hdfs://hadoop-test/tmp/output，其中，&hdfs://hadoop-test&是由Hadoop配置文件core- site.xml中参数fs.default.name指定的，具体替换成你的配置即可。
2. TopK编程实例
TopK程序的任务是对一堆文本进行词频统计，并返回出现频率最高的K个词。如果采用MapReduce实现，则需要编写两个作 业：WordCount和TopK，而使用Spark则只需一个作业，其中WordCount部分已由前面实现了，接下来顺着前面的实现，找到Top
K个词。注意，本文的实现并不是最优的，有很大改进空间。
步骤1：首先需要对所有词按照词频排序，如下：
val&sorted&=&result.map&{&&&case(key,&value)&=&(value,&key);&//exchange&key&and&value&}.sortByKey(true,&1)&
步骤2：返回前K个：
val&topK&=&sorted.top(args(3).toInt)&
步骤3：将K各词打印出来：
topK.foreach(println)&
注意，对于应用程序标准输出的内容，YARN将保存到Container的stdout日志中。在YARN中，每个Container存在三个日志 文件，分别是stdout、stderr和syslog，前两个保存的是标准输出产生的内容，第三个保存的是log4j打印的日志，通常只有第三个日志中 有内容。
本程序完整代码、编译好的jar包和运行脚本可以从这里下载。下载之后，按照&Apache Spark学习：利用Eclipse构建Spark集成开发环境&一文操作流程运行即可。
3. SparkJoin编程实例
在推荐领域有一个著名的开放测试集是movielens给的，下载链接是：http://grouplens.org/datasets/movielens/，该测试集包含三个文件，分别是ratings.dat、sers.dat、movies.dat，具体介绍可阅读：README.txt，本节给出的SparkJoin实例则通过连接ratings.dat和movies.dat两个文件得到平均得分超过4.0的电影列表，采用的数据集是：。程序代码如下：
import&org.apache.spark._&import&SparkContext._&object&SparkJoin&{&&&def&main(args:&Array[String])&{&&&&&if&(args.length&!=&4&){&&&&&&&println(&usage&is&org.test.WordCount&&master&&&rating&&&movie&&&output&&)&&&&&&&return&&&&&}&&&&&val&sc&=&new&SparkContext(args(0),&&WordCount&,&&&&&System.getenv(&SPARK_HOME&),&Seq(System.getenv(&SPARK_TEST_JAR&)))&&&&&&&&&&&&val&textFile&=&sc.textFile(args(1))&&&&&&&&&&&&val&rating&=&textFile.map(line&=&&{&&&&&&&&&val&fileds&=&line.split(&::&)&&&&&&&&&(fileds(1).toInt,&fileds(2).toDouble)&&&&&&&&})&&&&&&&val&movieScores&=&rating&&&&&&&&.groupByKey()&&&&&&&&.map(data&=&&{&&&&&&&&&&val&avg&=&data._2.sum&/&data._2.size&&&&&&&&&&(data._1,&avg)&&&&&&&&})&&&&&&&&&&&&&&val&movies&=&sc.textFile(args(2))&&&&&&val&movieskey&=&movies.map(line&=&&{&&&&&&&&val&fileds&=&line.split(&::&)&&&&&&&&&(fileds(0).toInt,&fileds(1))&&&&&&}).keyBy(tup&=&&tup._1)&&&&&&&&&&&&&&val&result&=&movieScores&&&&&&&&.keyBy(tup&=&&tup._1)&&&&&&&&.join(movieskey)&&&&&&&&.filter(f&=&&f._2._1._2&&&4.0)&&&&&&&&.map(f&=&&(f._1,&f._2._1._2,&f._2._2._2))&&&&&&&result.saveAsTextFile(args(3))&&&}&}&
你可以从代码、编译好的jar包和运行脚本。
这个程序直接使用Spark编写有些麻烦，可以直接在Shark上编写HQL实现，Shark是基于Spark的类似Hive的交互式查询引擎，具体可参考：。
Spark 程序设计对Scala语言的要求不高，正如Hadoop程序设计对Java语言要求不高一样，只要掌握了最基本的语法就能编写程序，且常见的语法和表达方式是很少的。通常，刚开始仿照官方实例编写程序，包括Scala、Java和Python三种语言实例。
原文链接：
【编辑推荐】
【责任编辑： TEL：（010）】
大家都在看猜你喜欢
专题头条原创头条原创
24H热文一周话题本月最赞
讲师：1人学习过
讲师：33人学习过
讲师：0人学习过
精选博文论坛热帖下载排行
本书是关于EJB 3.0的专业技术教程，专注于EJB的概念、方法、开发过程的介绍。同时，本书还研究许多高端的EJB知识，使得开发者能够真正理解...
订阅51CTO邮刊3289人阅读
分布式系统_spark（13）
一、编写第一个用scala写的spark应用：
仿照spark的 quick-start的Self-Contained Applications写出第一个scala完整程序
链接如下：
http://spark.apache.org/docs/latest/quick-start.html
/* SimpleApp.scala */
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
object SimpleApp {
def main(args: Array[String]) {
val logFile = &YOUR_SPARK_HOME/README.md& // Should be some file on your system
val conf = new SparkConf().setAppName(&Simple Application&)
val sc = new SparkContext(conf)
val logData = sc.textFile(logFile, 2).cache()
val numAs = logData.filter(line =& line.contains(&a&)).count()
val numBs = logData.filter(line =& line.contains(&b&)).count()
println(&Lines with a: %s, Lines with b: %s&.format(numAs, numBs))
整个程序作用是：找到这个文件
YOUR_SPARK_HOME/README.md
中有几个a和几个b。
二、用sbt进行打包成jar：
命令：sbt package
具体步骤见
http://spark.apache.org/docs/latest/quick-start.html
中的Self-Contained Applications
打包时候几点注意：
目录结构一定要对
目录结构可以通过find .来看
有点类似cmake的感觉
总时间，近30分钟，开始打开会terminal没现象10分钟，然后开始要各种resolve，之后要下载很多库，我这边网速超慢
sbt是个联网编译器，
spark的应用用到了很多RDD的变换，来编译这些库都得去网上下相应的包
最后显示，编译时间11s
这时成功完成SimpleApp
三、在本机上测试：
YOUR_SPARK_HOME/bin/spark-submit \
&--class &SimpleApp& \
&--master local[4] \
&target/scala-2.10/simple-project_2.10-1.0.jar
&我对jar的理解就是一个可执行文件了，这个可执行文件在JVM上就可以跑了，local中4是指设置成4个线程，但具体原因我也不知道
注意submit的参数：
--class中 SimpleApp是包名
上传的的jar的地址别写错
第二次编译不知道会怎么样，因为按理说这些库都下好了，下次就不需要再重新下载了吧，不清楚包的利用率高不高
&&相关文章推荐
参考知识库
* 以上用户言论只代表其个人观点，不代表CSDN网站的观点或立场
访问：26122次
排名：千里之外
原创：44篇
转载：17篇
(3)(1)(4)(3)(5)(17)(1)(4)(5)(5)(15)