如何根据匹配值对 PySpark 和 SQL 中的列值进行分组？

python

2024-06-04 07:00:46

根据匹配值对 PySpark 和 SQL 中的列值进行分组

问题陈述

在数据处理中，我们经常需要将数据按特定条件进行分组。一个常见的情况是根据两列中的匹配值对值进行分组。本文将演示如何使用 PySpark 和 SQL 来解决此问题。

PySpark 解决方案

1. 创建数据框

首先，我们创建一个包含要分组的数据的 Spark 数据框。

2. 连接列

接下来，我们将 CustomerCode 和 TanList 列连接成一个新列。

3. 分组和聚合

然后，我们按新列进行分组，并聚合 CustomerCode 和 TanList 列。

4. 分割和提取

最后，我们分割新列以提取原始 CustomerCode 和 TanList 值。

Python 代码：

import pyspark.sql.functions as F

df = spark.createDataFrame([
    ("4-1234", "MUMS12345A,BLRS12345E,BLRS12345G"),
    ("4-1235", "MUMS12345A,CHED12345A"),
    ("4-1236", "RTKD12345A")
], ["CustomerCode", "TanList"])

df = df.withColumn("ConcatValues", F.concat_ws(",", "CustomerCode", "TanList"))
df = df.groupBy("ConcatValues").agg(F.collect_list("CustomerCode").alias("CustomerCodeList"), F.collect_list("TanList").alias("TANList"))
df = df.withColumn("CustomerCodeList", F.regexp_extract("ConcatValues", "(.*?),(.*)", 1))
df = df.withColumn("TANList", F.regexp_extract("ConcatValues", "(.*?),(.*)", 2))
df = df.dropDuplicates()

df.show()

SQL 解决方案

1. 连接列

在 SQL 中，我们可以使用 CONCAT() 函数将 CustomerCode 和 TanList 连接成一个新列。

2. 分组和聚合

然后，我们按新列分组，并聚合 CustomerCode 和 TanList 列。

3. 分割和提取

最后，我们可以使用正则表达式或其他技术分割新列以提取原始 CustomerCode 和 TanList 值。

SQL 查询：

SELECT
    GROUP_CONCAT(DISTINCT CustomerCode) AS CustomerCodeList,
    GROUP_CONCAT(DISTINCT TanList) AS TANList
FROM (
    SELECT
        CustomerCode,
        TanList,
        CONCAT(CustomerCode, ",", TanList) AS ConcatValues
    FROM your_table
) AS t
GROUP BY ConcatValues