Spark编程快速入门

2024年2月5日 188次阅读来源: 追寻者的小憩书铺

本教程基于Spark官网的快速入门教程改编而来，官方文档和中文翻译的传送门见最下方。（注意，实际情况可能因为版本不同等原因有所不同）

如果对本文中的一些名词感到疑惑，可以看另外一篇入门文章：五分钟大数据：Spark入门

关键字：

使用 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的连接