PySpark 使用pyspark在Spark 2.0中构建sparkSession

在本文中，我们将介绍如何使用pyspark在Spark 2.0中构建sparkSession。在Spark 2.0及以上版本中，sparkSession成为了默认的入口点，可以用来创建DataFrame、注册表、执行SQL查询等。

什么是sparkSession？

SparkSession是与Spark集群通信的主要接口。在早期版本的Spark中，SparkContext是主要的入口点，但在Spark 2.0中，SparkSession被引入为更高级别的API，用于创建和管理DataFrame和Dataset。

SparkSession封装了Spark Context，因此通过SparkSession创建DataFrame比通过SparkContext创建DataFrame更为方便。SparkSession还提供了一些高级功能，如注册临时表、执行SQL查询等。

创建sparkSession

要在Spark 2.0中创建sparkSession，我们首先需要导入pyspark.sql模块，然后使用SparkSession.builder创建一个sparkSession实例。下面是一个简单的示例：

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("PySpark Example") \
    .config("spark.some.config.option", "some-value") \
    .getOrCreate()

# 从sparkSession创建DataFrame
df = spark.read.csv("input.csv", header=True, inferSchema=True)

在上面的示例中，我们使用SparkSession.builder创建一个sparkSession实例，并指定应用程序的名称和一些配置选项。getOrCreate方法将返回一个已经存在的sparkSession，如果不存在则创建一个新的sparkSession。

配置sparkSession

在创建sparkSession时，我们可以通过.config方法来配置选项。常用的配置选项包括：

appName：指定应用程序的名称；
master：指定Spark集群的URL；
spark.submit.deployMode：指定应用程序的部署模式，可以是client（在客户端运行）或cluster（在集群中运行）；
spark.some.config.option：其他特定的配置选项，如内存分配、并行度等。

下面的示例演示了如何配置sparkSession的一些常见选项：

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("PySpark Example") \
    .config("spark.some.config.option", "some-value") \
    .config("spark.memory.fraction", "0.8") \
    .config("spark.executor.instances", "4") \
    .getOrCreate()

在上面的示例中，我们通过.config方法设置了spark.some.config.option、spark.memory.fraction和spark.executor.instances这三个配置选项。

使用sparkSession创建DataFrame

通过sparkSession可以方便地创建DataFrame。DataFrame是基于RDD的分布式数据集，它提供了结构化数据处理的接口。

下面是一个使用sparkSession创建DataFrame的示例：

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("PySpark Example") \
    .getOrCreate()

# 从本地文件系统读取CSV文件
df = spark.read.csv("input.csv", header=True, inferSchema=True)

# 显示DataFrame的前几行
df.show(5)

# 使用SQL查询DataFrame
df.createOrReplaceTempView("my_table")
result = spark.sql("SELECT * FROM my_table WHERE age > 30")
result.show()

在上面的示例中，我们首先使用sparkSession读取了一个CSV文件，然后使用show方法显示了DataFrame的前5行。接下来，我们将DataFrame注册为临时表，并使用SQL查询过滤了年龄大于30的记录，并使用show方法显示了查询结果。

总结

本文介绍了如何使用pyspark在Spark 2.0中构建sparkSession。我们了解了sparkSession的概念和作用，并学习了创建sparkSession、配置sparkSession以及使用sparkSession创建DataFrame的方法。通过这些知识，我们可以更方便地使用pyspark进行数据处理和分析。