使用 Spark Shell 进行交互式分析

启动

Spark提供了一个学习API的简单方式，就是Spark shell——同时也是一个强大数据交互式分析工具。基于scala和python的Spark shell的启动方式如下：

Scala：./bin/spark-shell

Python：./bin/pyspark

注意，使用上面的方式登录，实现要把目录切换到Spark的安装目录下；如果已经配置好SPARK_HOME，那么也可以直接spark-shell或者pyspark启动

读取文件

Spark 的主要抽象是一个称为 Dataset 的分布式的 item 集合。Datasets 可以从 Hadoop 的 InputFormats（例如 HDFS文件）或者通过其它的 Datasets 转换来创建。

从外部读取文件

Scala：

scala>val textFile = spark.read.textFile("README.md")

Python：

>>> textFile = spark.read.text("README.md")

处理Dataset（1）：统计含有“Spark”的行数

Scala版：

scala> textFile.count() //计数

scala> textFile.first() //显示第一行

scala> val linesWithSpark = textFile.filter(line => line.contains("Spark")) //过滤出所需列，transform操作返回一个新的Dataset

scala> textFile.filter(line => line.contains("Spark")).count() //链式操作transform和action，返回所有含有"Spark"的行数

Python版：

>>> textFile.count()

>>> textFile.first()

>>> linesWithSpark = textFile.filter(textFile.value.contains("Spark"))

>>> textFile.filter(textFile.value.contains("Spark")).count()

处理Dataset（2）：统计一行中的单词最大数

Scala版

scala> textFile.map(line => line.split(" ").size).reduce((a, b) => if (a > b) a else b)

//方法2：

scala> import java.lang.Math

scala> textFile.map(line => line.split(" ").size).reduce((a, b) => Math.max(a, b))

Python版

>>> from pyspark.sql.functions import *

>>> textFile.select(size(split(textFile.value, "\s+")).name("numWords")).agg(max(col("numWords"))).collect()

处理Dataset（3）：实现mapreduce中的wordcount

Scala版

scala> val wordCounts = textFile.flatMap(line => line.split(" ")).groupByKey(identity).count()

scala> wordCounts.collect()

Python版

>>> wordCounts = textFile.select(explode(split(textFile.value, "\s+")).as("word")).groupBy("word").count()

>>> wordCounts.collect()

处理Dataset（4）：使用缓存

因为Spark是基于内存的运算，所以速度要比基于硬盘的MapReduce快很多；为了高效使用内存，Spark会将已经使用过的空间回收，但是有一些数据，我们想要重复使用，这时候我们就可以用到缓存技术，直接使用之前的计算结果。

Scala

scala> linesWithSpark.cache()

Python

>>> linesWithSpark.cache()

这里同时列出Scala和Python的实现方式，是为了大家更好的对比。需要注意的是，很多在Scala里的api，在python里都是不存在的，或者形式是完全不同的，因为python有自己的方法来处理。

独立的应用

学习完交互式探索之后，我们再来看一下独立的Spark应用如何去做。下面的demo的作用是统计一行中“a”“b”出现的次数。

代码编写

Scala版

/* SimpleApp.scala */

import org.apache.spark.sql.SparkSession

object SimpleApp {

def main(args: Array[String]) {

val logFile = "YOUR_SPARK_HOME/README.md"

val spark = SparkSession.builder.appName("Simple Application").getOrCreate() //首先需要建立与spark的连接

val logData = spark.read.textFile(logFile).cache()

val numAs = logData.filter(line => line.contains("a")).count()

val numBs = logData.filter(line => line.contains("b")).count()

println(s"Lines with a: $numAs, Lines with b: $numBs")

spark.stop() //最后需要停止与spark的连接

}

}

Python版

"""SimpleApp.py"""

from pyspark.sql import SparkSession

logFile = "YOUR_SPARK_HOME/README.md"

spark = SparkSession.builder().appName(appName).master(master).getOrCreate() #建立与spark的连接

logData = spark.read.text(logFile).cache()

numAs = logData.filter(logData.value.contains('a')).count()

numBs = logData.filter(logData.value.contains('b')).count()

print("Lines with a: %i, lines with b: %i" % (numAs, numBs))

spark.stop() #停止与spark的连接

任务提交

完成独立程序的编程之后，接下来就要将任务提交到spark来执行，执行的方法如下：

# Scala 使用 run-example:

./bin/run-example SparkPi

# Python 使用 spark-submit:

./bin/spark-submit examples/src/main/python/pi.py

文集

文章

补充资源

官方文档：http://spark.apache.org/docs/latest/quick-start.html

官方文档中文版：http://spark.apachecn.org/docs/cn/2.2.0/quick-start.html