按列中的相似名称对行进行分组

Question

对于计算，我已经创建了这个数据框

df1:

+--------------+------+------------+
|name          |MG    |questions   |
+--------------+------+------------+
|toto          |MG1   |[Q1, Q2]    |
|toto          |MG3   |[Q4, Q6, Q7]|
|toto-123      |MG1   |[Q1, Q2]    |
|toto-456      |MG1   |[Q1, Q2]    |
|titi          |MG1   |[Q1, Q2]    |
|...           |...   |...         |
+--------------+------+------------+

到目前为止，这已经足够了，但是现在，对于一个新的需求，我遇到了一个我无法解决的问题：我必须按 name 和 MG 而不是精确的 name 对行进行分组，但是有类似的name，就像那样

df1:

+--------------+------+-----------------------------+
|name          |MG    |questions                    |
+--------------+------+-----------------------------+
|toto          |MG1   |[Q1, Q2] [Q1, Q2] [Q1, Q2]   |
|toto          |MG3   |[Q4, Q6, Q7]                 |
|titi          |MG1   |[Q1, Q2]                     |
|...           |...   |...                          |
+--------------+------+-----------------------------+

这里，questions for toto 是toto, toto-123 and toto-456 的问题。它们可以保留在不同的阵列中或相同的阵列中。理想情况下，toto-123 和 toto-456 的行仍然保留。

我已尝试使用条件思考 SQL 表达式的 groupby 函数，但没有成功

您有什么提示或解决方案吗？ 非常感谢

Answer 1

您可以首先使用自联接查找具有相似name 和mg 值的行，在联接条件中使用like 运算符：

import pyspark.sql.functions as F

df = spark.createDataFrame([
    ("toto", "MG1", ["Q1", "Q2"]), ("toto", "MG3", ["Q4", "Q6", "Q7"]),
    ("toto-123", "MG1", ["Q1", "Q2"]), ("toto-456", "MG1", ["Q1", "Q2"]),
    ("titi", "MG1", ["Q1", "Q2"])], ["name", "mg", "questions"])

similar_df = df.alias("a").join(
    df.alias("b"),
    F.expr("a.name like concat('%', b.name, '%') and a.mg = b.mg and a.name != b.name"),
).selectExpr("a.name", "a.mg", "b.name as similar_name")

similar_df.show()
#+--------+---+------------+
#|    name| mg|similar_name|
#+--------+---+------------+
#|toto-123|MG1|        toto|
#|toto-456|MG1|        toto|
#+--------+---+------------+

然后，再次加入原始数据框，并将对应名称的值更改为第一步找到的相似名称，最后按name和mg分组，收集questions列表：

result = df.alias("df").join(
    similar_df.alias("s"),
    ["name", "mg"],
    'left'
).select(
    F.coalesce(F.col("similar_name"), F.col("df.name")).alias("name"),
    F.col("mg"),
    F.col("df.questions"),
).groupBy("name", "mg").agg(
    F.collect_list("questions").alias("questions")
)

result.show(truncate=False)
#+----+---+------------------------------+
#|name|mg |questions                     |
#+----+---+------------------------------+
#|toto|MG1|[[Q1, Q2], [Q1, Q2], [Q1, Q2]]|
#|titi|MG1|[[Q1, Q2]]                    |
#|toto|MG3|[[Q4, Q6, Q7]]                |
#+----+---+------------------------------+