PySpark中不使用explode函数对比不同DataFrame的String与Array<string>列

我有两个DataFrame：
df1：

sku   category   cep   seller   state
4858    BDU     00000   xefd     SP

df2：

depth  price   sku  seller  infos_product
6.1    5.60  47347   gaha   [{1, 86800000, 86...

df2的schema如下：

|-- depth: double (nullable = true)
 |-- sku: string (nullable = true)
 |-- price: double (nullable = true)
 |-- seller: string (nullable = true)
 |-- infos_produt: array (nullable = false)
 |    |-- element: struct (containsNull = false)
 |    |    |-- modality_id: integer (nullable = true)
 |    |    |-- cep_coleta_ini: integer (nullable = true)
 |    |    |-- cep_coleta_fim: integer (nullable = true)
 |    |    |-- cep_entrega_ini: integer (nullable = true)
 |    |    |-- cep_entrega_fim: integer (nullable = true)
 |    |    |-- cubage_factor_entrega: double (nullable = true)
 |    |    |-- value_coleta: double (nullable = true)
 |    |    |-- value_entrega: double (nullable = true)
 |    |    |-- city: string (nullable = true)
 |    |    |-- state: string (nullable = true)

我需要对这两个DataFrame进行关联检查，尝试执行如下代码：

condi = [(df1.seller_id == df2.seller) & (df2.infos_produt.state == df1.state)]

df_finish = (df1\
    .join(df2, on = condi ,how='left'))

但出现报错：

AnalysisException: cannot resolve '(infos_produt.`state` = view.coverage_state)' due to data type mismatch: differing types in '(infos_produt.`state` = view.coverage_state)' (array<string> and string).

因数据量较大，explode函数无法适用，希望能在不使用explode函数的前提下解决该问题。

报错核心是df2.infos_produt.state返回字符串数组（array），而df1.state是单个字符串，直接用等号比较会触发类型不匹配。以下是三种不依赖explode的解决方法：

方法1：用array_contains做包含判断 #

如果需求是只要infos_produt数组中存在任意一个struct的state等于df1.state就关联成功，用Spark内置的array_contains函数即可：

from pyspark.sql import functions as F

# 注意：原代码中df1.seller_id应为df1.seller（匹配df1的schema）
condi = [
    df1.seller == df2.seller,
    F.array_contains(F.col("infos_produt.state"), df1.state)
]

df_finish = df1.join(df2, on=condi, how='left')

array_contains直接检查数组是否包含目标元素，无需展开数组，性能高效。

方法2：用exists函数自定义判断（Spark 3.0+） #

如果需要更灵活的条件（比如同时校验多个字段），可以用exists遍历数组元素：

from pyspark.sql import functions as F

condi = [
    df1.seller == df2.seller,
    F.exists(
        "infos_produt",
        lambda elem: elem.state == df1.state
    )
]

df_finish = df1.join(df2, on=condi, how='left')

exists会对数组内每个struct执行lambda逻辑，只要有一个元素满足条件就返回True，支持复杂的多字段匹配场景。

方法3：提前过滤df2缩小数据集 #

针对超大规模数据场景，可以先过滤df2，只保留包含目标state的记录，再执行关联：

from pyspark.sql import functions as F

# 先提取df1中所有distinct的state值，过滤df2
target_states = df1.select("state").distinct().rdd.flatMap(lambda x: x).collect()
df2_filtered = df2.filter(
    F.exists(
        "infos_produt",
        lambda elem: elem.state.isin(target_states)
    )
)

# 再执行关联
condi = [
    df1.seller == df2_filtered.seller,
    F.array_contains(F.col("infos_produt.state"), df1.state)
]

df_finish = df1.join(df2_filtered, on=condi, how='left')

这种方式先减少df2的数据量，能大幅降低后续关联的计算开销。

内容的提问来源于stack exchange，提问作者Vivian