Pyspark：如何过滤 MapType 列上的数据框？ （如 isin() 的风格）

Question

当我想以isin() 的样式过滤 MapType 列上的数据框时，最好的策略是什么？

所以基本上我想获取数据框的所有行，其中 MapType 列的内容与 MapType-“实例”列表中的条目之一匹配。也可以加入该列，但到目前为止我尝试的所有方法都失败了，因为EqualTo does not support ordering on type map。

除了使用 isin() 或 join() 的直接方法之外，我还提出了使用 to_json() 将地图转储到 json 的想法，然后过滤 Json 字符串，但这似乎是随机排序的键，所以这个字符串比较也不可靠？

我错过了什么简单的东西吗？您建议如何解决这个问题？

例子df：

+----+---------------------------------------------------------+
|key |metric                                                   |
+----+---------------------------------------------------------+
|123k|Map(metric1 -> 1.3, metric2 -> 6.3, metric3 -> 7.6)      |
|d23d|Map(metric1 -> 1.5, metric2 -> 2.0, metric3 -> 2.2)      |
|as3d|Map(metric1 -> 2.2, metric2 -> 4.3, metric3 -> 9.0)      |
+----+---------------------------------------------------------+

过滤器（伪代码）：

df.where(metric.isin([
 Map(metric1 -> 1.3, metric2 -> 6.3, metric3 -> 7.6),
 Map(metric1 -> 1.5, metric2 -> 2.0, metric3 -> 2.2)
])

期望的输出：

----+---------------------------------------------------------+
|key |metric                                                   |
+----+---------------------------------------------------------+
|123k|Map(metric1 -> 1.3, metric2 -> 6.3, metric3 -> 7.6)      |
|d23d|Map(metric1 -> 1.5, metric2 -> 2.0, metric3 -> 2.2)      |
+----+---------------------------------------------------------+

Answer 1

比较 Spark 中的 2 个地图列并不那么明显。对于第一个映射中的每个键，您需要检查第二个映射中是否具有相同的值。键也一样。

使用 UDF 可能更简单，因为在 Python 中您可以检查 dict 相等性：

from pyspark.sql import functions as F

map_equals = F.udf(lambda x, y: x == y, BooleanType())

# create map1 literal to filter with
map1 = F.create_map(*[
    F.lit(x) for x in chain(*{"metric1": 1.3, "metric2": 6.3, "metric3": 7.6}.items())
])

df1 = df.filter(map_equals("metric", map1))

df1.show(truncate=False)

#+----+------------------------------------------------+
#|key |metric                                          |
#+----+------------------------------------------------+
#|123k|[metric1 -> 1.3, metric2 -> 6.3, metric3 -> 7.6]|
#+----+------------------------------------------------+

---

另一种方法是将要过滤的映射文字添加为列，并检查metric 中的每个键是否从该文字映射中获得相同的值。

这是一个在映射键数组上使用transfrom 和array_min 来创建过滤器表达式的示例。 （如果 array_min 返回 true，则表示所有值都相等）：

filter_map_literal = F.create_map(*[
    F.lit(x) for x in chain(*{"metric1": 1.3, "metric2": 6.3, "metric3": 7.6}.items())
])

df1 = df.withColumn("filter_map", filter_map_literal).filter(
    F.array_min(F.expr("""transform(map_keys(metric),
                           x -> if(filter_map[x] = metric[x], true, false)
                    )""")
                )
).drop("filter_map")

---

Answer 2

不是比较映射相等性的最优雅的方法：您可以收集映射键，比较两个映射中每个键的值，并确保所有值都相同。我想最好构造一个过滤器 df，然后进行半连接，而不是使用 isin 传递它们：

采样df和过滤df：

df.show(truncate=False)
+----+------------------------------------------------+
|key |metric                                          |
+----+------------------------------------------------+
|123k|[metric1 -> 1.3, metric2 -> 6.3, metric3 -> 7.6]|
|d23d|[metric1 -> 1.5, metric2 -> 2.0, metric3 -> 2.2]|
|as3d|[metric1 -> 2.2, metric2 -> 4.3, metric3 -> 9.0]|
+----+------------------------------------------------+

filter_df = df.select('metric').limit(2)
filter_df.show(truncate=False)
+------------------------------------------------+
|metric                                          |
+------------------------------------------------+
|[metric1 -> 1.3, metric2 -> 6.3, metric3 -> 7.6]|
|[metric1 -> 1.5, metric2 -> 2.0, metric3 -> 2.2]|
+------------------------------------------------+

过滤方式：

import pyspark.sql.functions as F

result = df.alias('df').join(
    filter_df.alias('filter_df'),
    F.expr("""
        aggregate(
            transform(
                concat(map_keys(df.metric), map_keys(filter_df.metric)),
                x -> filter_df.metric[x] = df.metric[x]
            ),
            true,
            (acc, x) -> acc and x
        )"""),
     'left_semi'
)

result.show(truncate=False)
+----+------------------------------------------------+
|key |metric                                          |
+----+------------------------------------------------+
|123k|[metric1 -> 1.3, metric2 -> 6.3, metric3 -> 7.6]|
|d23d|[metric1 -> 1.5, metric2 -> 2.0, metric3 -> 2.2]|
+----+------------------------------------------------+