看spark和scala版本

运行spark-shell

Welcome to

____ __

/ __/__ ___ _____/ /__

_\ \/ _ \/ _ `/ __/ '_/

/___/ .__/\_,_/_/ /_/\_\ version 1.6.0

/_/

 

Using Scala version 2.10.5 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_181)

得到spark版本1.6.0，scala版本2.10.5

 

下载安装特定版本的scala

https://www.scala-sbt.org/download.html

 

 

下载特定版本的spark包

https://archive.apache.org/dist/spark/

spark-1.6.0.tgz

其中examples是示例代码

 

开发

idea的scala插件安装参见Scala 写第一个程序HelloWorld：https://blog.csdn.net/whq12789/article/details/89453424。

idea新建Project，选择Scala——sbt，输入名称目录，注意版本选择与spark的scala相同版本。

build.sbt添加依赖，此处版本要求与spark版本相同

libraryDependencies += "org.apache.spark" %% "spark-sql" % "1.6.0"

src/main/scala右击新建Package：com.whq.test

右击包，新建Scala Class选择Kind为Object，名称为Hi，代码如下

package com.whq.testimport org.apache.spark.{SparkConf, SparkContext}object HdfsWordCount {def main(args: Array[String]) {if (args.length < 2) {System.err.println("Usage: HdfsWordCount <directory>")System.exit(1)}val inputFile=args(0)val outputFile=args(1)val conf = new SparkConf().setAppName("wordCount")// Create a Scala Spark Context.val sc = new SparkContext(conf)// Load our input data.val input =  sc.textFile(inputFile)// Split up into words.val words = input.flatMap(line => line.split(" "))// Transform into word and count.val counts = words.map(word => (word, 1)).reduceByKey{case (x, y) => x + y}// Save the word count back out to a text file, causing evaluation.counts.saveAsTextFile(outputFile)}}

统计分组信息。

右侧sbt点sbt tasks中的package，打包后的文件拷贝到服务器上准备运行。

 

su - hdfs

vi hello.txt添加内容

you,jumpi,jumpyou,jumpi,jumpjump

 

提交到spark目录下

hdfs dfs -put hello.txt /spark/

清空目标目录

hdfs dfs -rm -r /spark/out

 

提交运行

spark-submit --master yarn --deploy-mode client --class com.whq.test.HdfsWordCount sparksbt_2.10-0.1.jar /spark/hello.txt /spark/out

 

结果会输出到hdfs的/spark/out目录中

查看结果命令

hdfs dfs -cat /spark/out/_SUCCESS
hdfs dfs -cat /spark/out/part-00000
hdfs dfs -cat /spark/out/part-00001

 

 

Spark常用接口

常用接口

Spark主要使用到如下这几个类：

• SparkContext：是Spark的对外接口，负责向调用该类的java应用提供Spark的各种功能，如连接Spark集群，创建RDD等。
• SparkConf：Spark应用配置类，如设置应用名称，执行模式，executor内存等。
• RDD（Resilient Distributed Dataset）：用于在Spark应用程序中定义RDD的类。
• PairRDDFunctions：为key-value对的RDD数据提供运算操作，如groupByKey。
• Broadcast： 广播变量类。广播变量允许保留一个只读的变量，缓存在每一台机器上，而非每个任务保存一份拷贝。
• StorageLevel： 数据存储级别，有内存（MEMORY_ONLY），磁盘（DISK_ONLY），内存+磁盘（MEMORY_AND_DISK）等。

RDD上支持两种类型的操作: ：transformation和action，这两种类型的常用方法如下。

表 transformation

方法	说明
map(func)	对调用map的RDD数据集中的每个element都使用func。
filter(func)	对RDD中所有元素调用func，返回f为true的元素。
flatMap(func)	先对RDD所有元素调用func，然后将结果扁平化。
sample(withReplacement,faction,seed)	抽样。
union(otherDataset)	返回一个新的dataset，包含源dataset和给定dataset的元素的集合。
distinct([numTasks])	去除重复元素。
groupByKey(numTasks)	返回(K,Iterable[V])，将key相同的value组成一个集合。
reduceByKey(func,[numTasks])	对key相同的value调用func。
sortByKey([ascending],[numTasks])	按照key来进行排序，是升序还是降序，ascending是boolean类型。
join(otherDataset,[numTasks])	当有两个KV的dataset(K,V)和(K,W)，返回的是(K,(V,W))的dataset,numTasks为并发的任务数。
cogroup(otherDataset,[numTasks])	当有两个KV的dataset(K,V)和(K,W)，返回的是(K,Seq[V],Seq[W])的dataset,numTasks为并发的任务数。
cartesian(otherDataset)	笛卡尔积。

表 action

API	说明
reduce(func)	对RDD中的元素调用func。
collect()	返回包含RDD中所有元素的一个数组。
count()	返回的是dataset中的element的个数。
first()	返回的是dataset中的第一个元素。
take(n)	返回前n个elements。
takeSample(withReplacement，num，seed)	takeSample(withReplacement，num，seed)对dataset随机抽样，返回有num个元素组成的数组。withReplacement表示是否使用replacement。 。
saveAsTextFile（path）	把dataset写到一个text file中，或者hdfs，或者hdfs支持的文件系统中，spark把每条记录都转换为一行记录，然后写到file中。
saveAsSequenceFile(path)	只能用在key-value对上，然后生成SequenceFile写到本地或者hadoop文件系统。
countByKey()	对每个key出现的次数做统计。
foreach(func)	在数据集的每一个元素上，运行函数func。
countByValue()	对RDD中每个元素出现的次数进行统计。