Apache Spark

什么是Apache Spark?

Apache Spark是一个快速通用的大规模数据处理引擎

由Scala语言编写

运行在JVM中的函数式编程语言

Spark Shell

交互式的学习、数据探索,或特殊的分析

Python或Scala

Spark应用程序

大规模数据处理

Python、Scala或Java

Spark架构

Spark提供了一组基于Spark core上构建的函数库

Spark核心提供了基本的Spark抽象:弹性分布式数据集Resilient Distributed Datasets(RDD)

Spark SQL处理结构化数据

MLlib支持可扩展的机器学习

Spark Streaming应用处理实时数据

GraphX使用图表和图并行化计算

Spark SQL

Spark SQL是一个Spark库用于处理结构化数据

Spark SQL提供了什么?

DataFrame和Dataset的API

开发Spark应用程序的主要入口点

DataFrames和Datasets是结构化数据的抽象

一个SQL引擎和命令行接口

Spark Shell

Spark shell提供了一个交互式的Spark环境

通常称为REPL(读/评估/打印/循环)

用于学习、测试、数据探索,或特殊分析

可以使用Python或Scala运行Spark Shell

通常在gateway节点上运行Spark shell

开始使用Spark Shell

Spark Shell有很多不同的启动选项,包括

master:指定集群连接

jars:额外的JAR文件(仅Scala)

py-files:额外的Python文件(仅Python)

name:供UI使用的Spark应用程序名称

默认PySparkShell(Python)或Spark shell(Scala)

help:显示所有可用的shell选项

Spark集群选项

Spark应用程序可以运行在以下这些类型的集群上

Apache Hadoop YARN

Apache Mesos

Spark Standalone

也可以在本地运行,而不是在一个集群

Spark的默认集群类型为YARN

指定集群的类型或URL要使用master选项

master选项的可能值为

yarn

spark://masternode:port(Spark Standalone)

mesos://masternode:port(mesos)

local[*]在本地运行尽可能多的线程核心(默认)

local[n]在本地运行n线程

local用单个线程在本地运行

Spark Session

Spark API的主要入口是一个Spark Session

Spark shell提供了一个预配置的SparkSession对象，称为Spark

DataFrames和Datasets

DataFrames和Datasets是Spark中的主要数据表现形式

DataFrames表示表格形式的结构化数据

DataFrames模型数据类似于RDBMS表

DataFrames包含一个松散类型的行对象

行被组织成列所描述的模式

Datasets数据表示为一个指定类型的对象的集合

Datasets是编译时执行强类型检查,而不是运行时

将一个关联的模式对象映射到一个类似于表的行和列的结构

Datasets仅在Scala和Java语言中定义

DataFrame是Dataset[Row]的别名

DataFrames和行

DataFrames包含一个行对象的有序集合

行包含值的有序集合

行值可以是基本类型(如整数、字符串、和浮点数)或集合的类型(例如数组和列表)

模式列名称和类型映射到一行的值内

例子：创建DataFrame

users.json文本文件包含示例数据

每一行包含一条JSON记录，name,age,postal code

{"name":"Alice", "pcode":"94304"}

{"name":"Brayden", "age":30, "pcode":"94304"}

{"name":"Carla", "age":19, "pcode":"10036"}

{"name":"Diana", "age":46}

{"name":"Etienne", "pcode":"94104"}

使用spark.read创建一个DataFrame

返回Spark session的DataFrameReader

调用json函数来创建一个性的DataFrame

val usersDF = spark.read.json("users.json")

DataFrames总是含有一个关联性的schema

DataFrameReader可以从数据推断出schema

使用printSchema来显示DataFrame的schema

> val usersDF = spark.read.json("users.json")

> usersDF.printSchema()

root

|-- age: long (nullable = true)

|-- name: string (nullable = true)

|-- pcode: string (nullable = true)

show方法以表格格式显示前几行

> val usersDF = spark.read.json("users.json")

> usersDF.printSchema()

root

|-- age: long (nullable = true)

|-- name: string (nullable = true)

|-- pcode: string (nullable = true)

> usersDF.show()

+----+-------+-----+

| age| name|pcode|

+----+-------+-----+

|null| Alice|94304|

| 30|Brayden|94304|

| 19| Carla|10036|

| 46| Diana| null|

|null|Etienne|94104|

+----+-------+-----+

DataFrame操作

有两种主要的DataFrame操作类型

转换操作基于现有的数据类型创建一个新的DataFrame

转换操作由应用程序的executor并行执行

行动操作从DataFrame中输出数据值

输出通常从executor返回到主Spark程序(称为驱动程序)或保存到文件中

行动操作

常见的DataFrame行动操作包括

count:返回行数

first:返回第一行(head()的同义词)

take(n):将前n行作为一个数组返回(head(n)的同义词)

show(n):显示的前n行以表格形式(默认值是20行)

collect:返回DataFrame中所有的行作为一个数组

write:将数据保存到文件或其他数据源

例子：take操作

> val usersDF = spark.read.json("users.json")

> val users = usersDF.take(3)

users: Array[org.apache.spark.sql.Row] =

Array([null,Alice,94304],

[30,Brayden,94304],

[19,Carla,10036])

转换操作

转换操作基于现有的一个DataFrame创建一个新的DataFrame

新的DataFrame可能有相同的schema或不同的

转换操作不返回任何值或数据给驱动器

数据仍然分布在应用程序的executor

DataFrames是不可变的

DataFrame中数据从不修改

使用转换操创建一个新的包含你所需要数据的DataFrame

常见的转换操作有

select:只包含指定的列

where:仅包括符合指定的表达式是true的行(filter的同义词)

orderBy:行按指定的列排序(sort的同义词)

join:连接两个DataFrames指定的列

limit(n):创建一个新的DataFrame只有前n行

例子：select和where操作

> val nameAgeDF = usersDF.select("name","age")

> nameAgeDF.show()

+-------+----+

| name| age|

+-------+----+

| Alice|null|

|Brayden| 30|

| Carla| 19|

| Diana| 46|

|Etienne|null|

+-------+----+

> val over20DF = usersDF.where("age > 20")

> over20DF.show()

+---+-------+-----+

|age| name|pcode|

+---+-------+-----+

| 30|Brayden|94304|

| 46| Diana| null|

+---+-------+-----+

> val nameAgeDF = usersDF.select("name","age")

> val nameAgeOver20DF = nameAgeDF.where("age > 20")

> nameAgeOver20DF.show()

+---+-------+

|age| name|

+---+-------+

| 30|Brayden|

| 46| Diana|

+---+-------+

一个查询中的转换操作可以连接在一起

> nameAgeDF = usersDF.select("name","age")

> nameAgeOver20DF = nameAgeDF.where("age > 20")

> nameAgeOver20DF.show()

等同于

> usersDF.select("name","age").where("age > 20").show()