PySpark 数据重塑/透视

在本文中，我们将介绍如何在PySpark中重塑/透视数据。数据重塑是一种数据转换技术，用于将数据从一种形式转换为另一种形式，以便更好地满足我们的分析和计算需求。在PySpark中，我们可以使用RDD和DataFrames来实现数据重塑操作。

阅读更多：PySpark 教程

RDD中的数据重塑

在PySpark中，RDD是一种弹性的分布式数据集，我们可以使用它来进行数据重塑。下面是一些常见的数据重塑操作：

1. flatMap操作

flatMap操作可以将每个输入元素映射为多个输出元素，然后将所有的输出元素合并为一个单一的RDD。下面是一个示例，展示如何使用flatMap操作将一行文本拆分为单词：

lines = spark.sparkContext.textFile("data.txt")
words = lines.flatMap(lambda line: line.split(" "))

2. map操作

map操作可以将每个输入元素映射为一个输出元素，并将所有的输出元素合并为一个新的RDD。下面是一个示例，展示如何使用map操作将每个数字加倍：

nums = spark.sparkContext.parallelize([1, 2, 3, 4, 5])
doubled_nums = nums.map(lambda x: x * 2)

3. filter操作

filter操作可以根据指定的条件筛选出符合条件的元素，然后将这些元素组成一个新的RDD。下面是一个示例，展示如何使用filter操作筛选出偶数：

nums = spark.sparkContext.parallelize([1, 2, 3, 4, 5])
even_nums = nums.filter(lambda x: x % 2 == 0)

4. reduceByKey操作

reduceByKey操作可以根据键将具有相同键的元素的值进行合并。下面是一个示例，展示如何使用reduceByKey操作对每个键的值求和：

data = [("key1", 1), ("key2", 2), ("key1", 3), ("key2", 4)]
rdd = spark.sparkContext.parallelize(data)
sums = rdd.reduceByKey(lambda a, b: a + b)

5. groupByKey操作

groupByKey操作根据键对元素进行分组，并将相同键的所有值组合成一个迭代器。下面是一个示例，展示如何使用groupByKey操作按键进行分组：

data = [("key1", 1), ("key2", 2), ("key1", 3), ("key2", 4)]
rdd = spark.sparkContext.parallelize(data)
grouped = rdd.groupByKey()

6. sortByKey操作

sortByKey操作根据键对元素进行排序。下面是一个示例，展示如何使用sortByKey操作按键进行排序：

data = [("key1", 1), ("key2", 2), ("key1", 3), ("key2", 4)]
rdd = spark.sparkContext.parallelize(data)
sorted_rdd = rdd.sortByKey()

DataFrame中的数据重塑

除了使用RDD进行数据重塑外，我们还可以使用PySpark的DataFrame进行数据重塑。PySpark的DataFrame提供了更高级的API，可以简化许多数据重塑操作。

1. pivot操作

pivot操作可以根据指定的列来旋转数据。它会将指定列的唯一值作为新的列名，并使用其他列的值填充新的列。下面是一个示例，展示如何使用pivot操作来透视数据：

df = spark.read.csv("data.csv", header=True, inferSchema=True)
pivot_df = df.groupBy("category").pivot("key").sum("value")

2. stack和unstack操作

stack操作可以将DataFrame的列旋转为行，而unstack操作可以将行旋转为列。下面是一个示例，展示如何使用stack和unstack操作进行数据重塑：

df = spark.createDataFrame([(1, "A", 100), (2, "B", 200), (3, "C", 300)], ["id", "category", "value"])
stacked_df = df.selectExpr("id", "(category, value) as data").select("id", "data.category", "data.value")
unstacked_df = stacked_df.groupBy("id").pivot("category").sum("value")

3. melt操作

melt操作可以将宽格式的DataFrame转换为长格式。宽格式表示每个实体的多个属性都在一行上，而长格式则表示每个实体的每个属性都在一行上。下面是一个示例，展示如何使用melt操作进行数据重塑：

df = spark.createDataFrame([(1, "A", 100, 200), (2, "B", 300, 400)], ["id", "category", "value1", "value2"])
melted_df = df.selectExpr("id", "stack(2, 'value1', value1, 'value2', value2) as (category, value)").select("id", "category", "value")