towardsdatascience.com/pyspark-explained-the-explode-and-collect-list-functions-834f45ff5ac5

PySpark SQL，Apache PySpark 中 SQL 的 Python 接口，是一套强大的工具，用于数据转换和分析。它旨在模拟数据库 SQL 系统中最常见的操作类型，Pyspark SQL 还能够利用 Spark 中可用的 dataframe 模式来提供额外的功能。

简而言之，Pyspark SQL 提供了一组丰富的函数，使开发者能够高效地操作和处理数据。

在这些函数中，我想特别强调的两个不太为人所知的函数，它们在以独特方式转换和聚合数据方面的能力尤为突出。这两个函数是 explode 和 collect_list 操作符。

在这篇文章中，我将详细解释每个函数的功能，并展示每个函数的使用案例和示例 PySpark 代码。

Explode

PySpark SQL 中的 explode 函数是一个多功能的工具，用于转换和展平嵌套数据结构，如数组或映射，为单个行。当处理包含嵌套集合的复杂数据集时，这个函数特别有用，因为它允许你分析并操作这些结构中的单个元素。

Pyspark 中的数组与其他计算机语言中的数组非常相似——即一种数据结构，它以特定的顺序在连续的内存位置中存储一系列元素，通常为同一类型。

Spark 中的映射在语言（如 Python）中相当于字典。它们包含一系列键值对，并且对于快速查找特定键的值非常有用。我们将在稍后看到使用 Explode 的示例。

当应用于数组列时，explode 函数为数组中的每个元素创建一行新行，并将元素值存储在一个新列中。默认情况下，这个新列被命名为 col，但你可以使用别名指定一个自定义列名。

类似地，当应用于映射列时，explode 函数创建两个新列：一个用于键，另一个用于值。默认情况下，这些列分别被命名为 key 和 value，但同样，你可以使用别名提供自定义列名。

Collect_list

PySpark SQL 中的 collect_list 函数是一个聚合函数，它从列中收集值并将它们转换为数组。当你需要重建或聚合使用其他 PySpark SQL 函数（如 explode）扁平化或转换的数据时，它特别有用。在许多方面，它可以被视为 explode 的补充函数。

这个函数通常与 groupBy 操作符结合使用，在聚合值之前根据一个或多个列对数据进行分组。

访问免费的 PySpark 开发环境

如果你想跟随本文中的代码，你需要访问一个 PySpark 开发环境。

如果你很幸运，可以通过工作、云服务或本地安装来访问 PySpark，那么请继续使用。如果没有，请查看下面的链接，我在那里详细介绍了如何访问一个名为 Databricks Community Edition 的优秀的免费在线 PySpark 开发环境。

Databricks 是一个基于云的数据工程、机器学习和分析平台，它围绕 Apache Spark 构建，为处理大数据工作负载提供了一个统一的环境。Databricks 的创始人创建了 Spark，所以他们很了解他们的产品。

如何访问免费的在线 Spark 开发环境

示例用例

现在我们对 explode 和 collect_list 能做什么有了更多的了解，让我们考虑一些它们的用例。

explode 函数

我们将开始使用 explode 函数来转换一个数组。如果你还记得，在 Spark 中，数组是一种存储相同类型元素固定大小顺序集合的数据结构。

我们将设置一个 PySpark dataframe，它包含一个文本列中的人名和他们在数组列中喜欢吃的果子的名称。

from pyspark.sql import SparkSession
from pyspark.sql.functions import explode

# Initialize Spark session
spark = SparkSession.builder.appName("ArrayExplodeExample").getOrCreate()

# Create a DataFrame with an array column - "fruits"
data = [
    ("John", ["apple", "banana", "cherry"]),
    ("Mary", ["orange", "grape"]),
    ("Jane", ["strawberry", "blueberry", "raspberry"]),
    ("Mark", ["watermelon"])
]

# Define schema and create DataFrame
df = spark.createDataFrame(data, ["name", "fruits"])

# Show the original DataFrame
df.show(truncate=False)

+----+----------------------------------+
|John|[apple, banana, cherry]           |
|Mary|[orange, grape]                   |
|Jane|[strawberry, blueberry, raspberry]|
|Mark|[watermelon]                      |
+----+----------------------------------+

对于许多需要分析这些数据的情况，如果 “name” 和 “fruit” 的每个不同组合都在单独的记录上，事情会变得容易得多。我们可以使用 explode 函数来实现这一点。

# Use explode function to flatten the array column
exploded_df = df.withColumn("fruit", explode(df.fruits))

# Show the exploded DataFrame
exploded_df["name","fruit"].show(truncate=False)

+----+----------+
|name|fruit     |
+----+----------+
|John|apple     |
|John|banana    |
|John|cherry    |
|Mary|orange    |
|Mary|grape     |
|Jane|strawberry|
|Jane|blueberry |
|Jane|raspberry |
|Mark|watermelon|
+----+----------+

现在数据看起来更像一个常规的数据表，如果我们想对它进行额外的 dataframe 或 SQL 操作以进行进一步分析，它现在组织得更好。

使用 explode 在 PySpark 中处理映射非常相似。

from pyspark.sql import SparkSession
from pyspark.sql.functions import explode, create_map, lit, col
from pyspark.sql.types import MapType, StringType

# Initialize Spark session
spark = SparkSession.builder.appName("ExplodeExample").getOrCreate()

# Sample data
data = [
    ("Tom", {"Salary": "£5000", "Bonus": "£0"}),
    ("Dick", {"Salary": "£2690", "Bonus": None}),
    ("Harry", {"Salary": "£45000", "Bonus": "£20000"})
]

# Create DataFrame
df = spark.createDataFrame(data, ["Name", "Remuneration"])

# Show original DataFrame
df.show(truncate=False)

+-------+-----------------------------------+
|Name   |Remuneration                       |
+-------+-----------------------------------+
|Tom    |{Salary -> £5000, Bonus -> £0}     |
|Dick   |{Salary -> £2690, Bonus -> null}   |
|Harry  |{Salary -> £45000, Bonus -> £20000}|
+-------+-----------------------------------+

应用 explode，这次我们将键值对拆分成了单独的记录。就像我们上一个例子一样，这为进一步分析提供了更好的组织。

remuneration_exploded = df.select(
    col("Name"),
    explode(col("Remuneration")).alias("key", "value")
)

# Show the transformed DataFrame
remuneration_exploded.show(truncate=False)

+-------+------+-------+
| Name  | key  | value |
+-------+------+-------+
|Tom    |Salary|£5000  |
|Tom    |Bonus |   £0  |
|Dick   |Salary|£2690  |
|Dick   |Bonus | null  |
|Harry  |Salary|£45000 |
|Harry  |Bonus |£20000 |
+-------+------+-------+

使用一个稍微复杂一点的例子来结束，假设我们有一个以下的 PySpark dataframe。

+----+-----------+-----------+
|col1|     col2  |     col3  |
+----+-----------+-----------+
| a  | [1, 2, 3] | [4, 5, 6] |
+----+-----------+-----------+

我们希望得到以下输出。

+------+-----+-------+
|col1  |col2  |col3  |
+------+------+------+
|   a  |   1  |   4  |
|   a  |   2  |   5  |
|   a  |   3  |   6  |
+------+------+------+

这比看起来要复杂。首先，让我们创建我们的输入测试数据。

testData = [('a',[1,2,3],[4,5,6]),]

df = spark.createDataFrame(data=testData, schema = ['col1','col2','col3'])

乍一看，你可能认为可以直接 explode col2 和 col3，但这不会起作用，因为你一次只能展开一个列。让我们试试，你就会明白我的意思。

df.select ("col1",explode("col2").alias("col2"),"col3").select("col1","col2",explode("col3").alias("col3")).show()

+----+----+----+
|col1|col2|col3|
+----+----+----+
|   a|   1|   4|
|   a|   1|   5|
|   a|   1|   6|
|   a|   2|   4|
|   a|   2|   5|
|   a|   2|   6|
|   a|   3|   4|
|   a|   3|   5|
|   a|   3|   6|
+----+----+----+

嗯，这不是我们想要的。为了做到我们想要的，我们必须有一个中间步骤，对于 PySpark 数组来说，它相当于 Python zip 操作。回想一下，Python zip 操作符接受两个可迭代对象并将它们拼接在一起。例如，如果我们有，

numbers = [1, 2, 3]

letters = ['a', 'b', 'c']

zipped=zip(numbers,letters)

print(list(zipped))

[(1, 'a'), (2, 'b'), (3, 'c')]

对数组的等效命令方便地称为 arrays_zip。因此，我们必须首先使用它来 “缝合” 我们数组和然后进行 explode。这可以通过 Pyspark SQL 或 dataframe 操作来实现。这是 SQL 中的解决方案。

from pyspark.sql.functions import *

#create a database table of our input data

df.createOrReplaceTempView("test_table")

spark.sql("select col1,tmp.col2,tmp.col3 from (select col1,explode(tmp) as tmp from (select col1,arrays_zip(col2,col3) as tmp from test_table))").show()

+------+------+------+
|col1  |col2  |col3  |
+------+------+------+
|   a  |   1  |   4  |
|   a  |   2  |   5  |
|   a  |   3  |   6  |
+------+------+------+

collect_list 函数

collect_list 函数接受一个 PySpark dataframe，该 dataframe 以记录为基础存储数据，并返回一个包含该数据的单个 dataframe 列作为集合。从这个意义上讲，它与 explode 函数的作用相反。一个快速示例将展示我的意思。假设我们有这个输入数据集，

testData = (['a'],['b'],['c']) 

df = spark.createDataFrame(data=testData, schema = ['letter_column']) 

df.printSchema() 

df.show() 

+-------------+ 
|letter_column| 
+-------------+ 
|            a| 
|            b| 
|            c| 
+-------------+ 

将 collect_list 应用到我们得到的数据上，

from pyspark.sql.functions import collect_list

df.select(collect_list("letter_column").alias("letter_row")).show()

+----------+  
|letter_row|  
+----------+  
| [a, b, c]| 
+----------+ 

通常我们不仅仅处理一列数据，所以，对于更复杂的问题，考虑我们有以下 PySpark dataframe，显示了 3 天内天然气和电力的批发价格。

+-----------+----------+------+ 
|Fuel       |      Date| Price| 
+-----------+----------+------+ 
|Gas        |2019-10-11|121.56| 
|Gas        |2019-10-10|120.56| 
|Electricity|2019-10-11|100.00| 
|Gas        |2019-10-12|119.56| 
|Electricity|2019-10-10| 99.00| 
|Electricity|2019-10-12|101.00| 
+-----------+----------+------+

我们希望以以下格式返回一个新的数据集。需要注意的是，从左到右的每种燃料价格应按日期顺序排列。

 +-----------+------------------------+
|Fuel       |Price_hist              |
+-----------+------------------------+
|Electricity|[99.0, 100.0, 101.0]    |
|Gas        |[120.56, 121.56, 119.56]|
+-----------+------------------------+

我们将首先编写一些代码来创建我们的输入数据集。

data = [
    ("Gas", "2019-10-11", 121.56),
    ("Gas", "2019-10-10", 120.56),
    ("Electricity", "2019-10-11", 100.00),
    ("Gas", "2019-10-12", 119.56),
    ("Electricity", "2019-10-10", 99.00),
    ("Electricity", "2019-10-12", 101.00)
]

# Create DataFrame
df = spark.createDataFrame(data, ["Fuel", "Date", "Price"])

# Show DataFrame
df.show()

现在，运行我们的代码，

from pyspark.sql.functions import collect_list

df.select("Fuel",collect_list("Price").alias("Price Hist")).show(truncate=False)

返回错误，

...
...
AnalysisException: [MISSING_GROUP_BY] The query does not include a GROUP BY clause. Add GROUP BY or turn it into the window functions using OVER clauses.;
Aggregate [Fuel#2, collect_list(Price#4, 0, 0) AS Price Hist#22]
+- LogicalRDD [Fuel#2, Date#3, Price#4], false

这并不好，很明显，我们不得不对燃料名称进行某种分组。幸运的是，collect_list 函数实际上是一个聚合函数，因此我们可以使用 agg 和 groupBy 操作，以及在对 dataframe 进行预排序操作后，得到我们想要的结果。运行这个，

from pyspark.sql.functions import collect_list

# Sort by Date to ensure prices are in date order
sorted_df = df.sort("Fuel", "Date")

# Group by Stock and collect prices into a list
result_df = sorted_df.groupBy("Fuel").agg(collect_list("Price").alias("Price_hist"))

# Show the result DataFrame
result_df.show(truncate=False)

给我们提供了所需输出。

+-----------+------------------------+
|Fuel       |Price_hist              |
+-----------+------------------------+
|Electricity|[99.0, 100.0, 101.0]    |
|Gas        |[120.56, 121.56, 119.56]|
+-----------+------------------------+

摘要

在这篇文章中，我介绍了 PySpark SQL 中两个较为不寻常的数据操作函数，并给出了一些它们可以非常有用的用例。

如果您需要将 dataframe 中的数组或字典数据字段转换为它们的组成部分，并将它们放入 dataframe 中的单独记录中，请使用 explode 函数。

可以将 collect_list 函数视为 explode 函数的逆函数。使用这个函数将单个 dataframe 记录中的项目聚合到集合中。

_ 好的，就到这里吧。我希望您觉得这篇文章有用。如果您觉得有用，请访问我的个人资料页面这个链接。从那里，您可以查看我发布的其他故事并订阅以获取我发布新内容的通知。_

我知道现在经济困难，钱包紧张，但如果您从这个文章中获得了真正的价值，请考虑*买给我一点小礼物。

如果您喜欢这个内容，我想您也会觉得这些文章很有趣。

SQL Explained: Common Table Expressions

Python on Steroids: The Numba Boost