写在前面:实际工作中其实不需要自己安装和配置,更重要的是会用。所以就不研究怎么安装配置了。
前面介绍过:简单来说,Spark是一款分布式的计算框架,用于调度成百上千的服务器集群,计算TB、PB乃至EB级别的海量数据。Spark作为全球顶级的分布式计算框架,支持众多的编程语言进行开发。Python语言,则是Spark重点支持的方向,体现为Python第三方库:PySpark。
一、快速入门
PySpark 应用程序从初始化开始,SparkSession这是 PySpark 的入口点。如下:
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
刚接触Pyspark你可能有点懵,为啥你看别的教程可能发现:
使用Pyspark首先需要构建一个执行环境入口对象,PySpark的执行环境入口对象是类SparkContext
1.1. SparkSession 和 SparkContext 的区别
在Apache Spark中,SparkSession和SparkContext是两个核心的概念,他们在不同版本的Spark使用。
SparkContext
SparkContext是老一代API的核心入口点,它代表了一个到Spark集群的连接。通过SparkContext,你可以访问分布式计算的所有功能,比如创建RDD(Resilient Distributed Dataset),执行任务等。它是与集群交互的主要方式,在Spark应用程序中扮演着至关重要的角色。每个JVM中只能有一个活跃的SparkContext。在早期版本的Spark中,开发者直接使用SparkContext来编写程序。
SparkSession
SparkSession是在Spark 2.0中引入的一个新的抽象层,旨在为开发者提供一个统一的切入点来使用Spark的所有功能,包括DataFrame API、SQL查询、流处理等等。SparkSession内部包含了SparkContext,同时也提供了更高级的功能,使得操作更加简便。它允许你无缝地在不同类型的Spark API之间切换,而不需要显式地管理底层的SparkContext。
- 统一性:SparkSession提供了一种统一的方式来访问Spark的各种功能,避免了需要单独初始化SQLContext、HiveContext以及SparkContext的情况。
- 易用性:对于新用户来说,SparkSession更加直观易用,因为它简化了很多配置和初始化过程。
- 功能性:除了包含原有的SparkContext功能外,还增加了对DataFrame和Dataset的支持,可以直接运行SQL查询,支持流处理等更多高级特性。
总结来说,如果你正在使用的是较新版本的Spark,推荐使用SparkSession作为你的主要编程入口,因为它不仅涵盖了SparkContext的所有功能,而且还提供了额外的高级特性,使得开发更加高效便捷。而对于那些仍在维护基于旧版Spark的应用程序的开发者来说,理解并使用SparkContext仍然是必要的。
二、DataFrame 创建
通常, pyspark.sql.SparkSession.createDataFrame
通过传递列表、元组、字典和pyspark.sql.Row 的列表、由此类列表组成的 RDD 来创建 PySpark DataFrame。
pyspark.sql.SparkSession.createDataFrame使用schema参数来指定 DataFrame 的数据类型。当省略时,PySpark 通过从数据中抽取样本来推断相应的数据类型。
- eg1:从行列表中创建一个不指明数据类型的 PySpark DataFrame
from datetime import datetime, date
import pandas as pd
from pyspark.sql import Row
df = spark.createDataFrame([
Row(a=1, b=2., c='string1', d=date(2000, 1, 1), e=datetime(2000, 1, 1, 12, 0)),
Row(a=2, b=3., c='string2', d=date(2000, 2, 1), e=datetime(2000, 1, 2, 12, 0)),
Row(a=4, b=5., c='string3', d=date(2000, 3, 1), e=datetime(2000, 1, 3, 12, 0))
])
df
------------------------------------------------------------------------------------------
out:DataFrame[a:bigint,b:double,c:字符串,d:日期,e:时间戳]
- eg2:创建具有明确数据类型的 PySpark DataFrame。
df = spark.createDataFrame([
(1, 2., 'string1', date(2000, 1, 1), datetime(2000, 1, 1, 12, 0)),
(2, 3., 'string2', date(2000, 2, 1), datetime(2000, 1, 2, 12, 0)),
(3, 4., 'string3', date(2000, 3, 1), datetime(2000, 1, 3, 12, 0))
], schema='a long, b double, c string, d date, e timestamp')
df
------------------------------------------------------------------------------------------
out:DataFrame[a:bigint,b:double,c:字符串,d:日期,e:时间戳]
- eg3:直接用pandas 的 DataFrame 创建 PySpark DataFrame
pandas_df = pd.DataFrame({
'a': [1, 2, 3],
'b': [2., 3., 4.],
'c': ['string1', 'string2', 'string3'],
'd': [date(2000, 1, 1), date(2000, 2, 1), date(2000, 3, 1)],
'e': [datetime(2000, 1, 1, 12, 0), datetime(2000, 1, 2, 12, 0), datetime(2000, 1, 3, 12, 0)]
})
df = spark.createDataFrame(pandas_df)
df
------------------------------------------------------------------------------------------
out:DataFrame[a:bigint,b:double,c:字符串,d:日期,e:时间戳]
- 上面创建的 DataFrames 都具有相同的结果和模式。
# All DataFrames above result same.
df.show()
df.printSchema()
------------------------------------out------------------------------------------------------
+---+---+-------+----------+-------------------+
| a| b| c| d| e|
+---+----+----------+----------+-------------------+
| 1|2.0|字符串1|2000-01-01|2000-01-01 12:00:00|
| 2|3.0|字符串2|2000-02-01|2000-01-02 12:00:00|
| 3|4.0|字符串3|2000-03-01|2000-01-03 12:00:00|
+---+---+-------+----------+-------------------+
root
|-- a: long(可空 = true)
|-- b: double(可空 = true)
|-- c: 字符串(可空 = true)
|-- d: 日期(可空 = true)
|-- e: 时间戳(可空 = true)
- 当创建了一个PySpark DataFrame,你可以通过如下语法查看其基本的数据结构和数据类型。
df.show(1)
df.show(1, vertical=True) # 行也可以垂直显示。当行太长而无法水平显示时,这很有用。
df.columns
df.select("a", "b", "c").describe().show() # 显示 DataFrame 的摘要
DataFrame.collect()
将分布式数据作为 Python 中的本地数据收集到驱动程序端。方便进行各种运算操作。
df.collect()
[out]:
[行(a=1,b=2.0,c='string1',d=datetime.date(2000, 1, 1),e=datetime.datetime(2000, 1, 1, 12, 0)),
行(a=2,b=3.0,c='string2',d=datetime.date(2000, 2, 1),e=datetime.datetime(2000, 1, 2, 12, 0)),
行(a=3,b=4.0,c='string3',d=datetime.date(2000, 3, 1),e=datetime.datetime(2000, 1, 3, 12, 0))]
- 为了避免引发内存不足异常,请使用DataFrame.take()或DataFrame.tail()。
df.take(1)
[out]:
[Row(a=1,b=2.0,c='string1',d=datetime.date(2000, 1, 1),e=datetime.datetime(2000, 1, 1, 12, 0))]
- PySpark DataFrame 还提供转换回pandas DataFrame 的功能,以利用 pandas API。
df.toPandas()