Databricks中PySpark按品牌关联查询文本的可行方案求助

在Databricks Notebook中，有如下结构的PySpark DataFrame（每行按城市唯一）：

+-------+--------------------------+------+----------+
|city   |brand                     |weight|date      |
+-------+--------------------------+------+----------+
|Dallas |['BMW', 'Ford', 'Chevy']  |0.94  |2021-05-05|
|Chicago|['Ford', 'VW', 'Toyota']  |0.92  |2021-05-07|
|Boston |['BMW', 'Toyota', 'Chevy']|0.78  |2021-05-08|
|Atlanta|['Toyota', 'VW', 'Chevy'] |0.83  |2021-05-09|
|Phoenix|['BMW', 'Honda', 'Toyota']|0.89  |2021-05-12|
+-------+--------------------------+------+----------+

需求：对每行的品牌列表中的每个品牌，结合该行的城市、日期查询Databricks上的目标表，获取对应文本后拼接，输出格式不限。

此前尝试在UDF中执行spark.sql(sql_stmt)时，抛出错误：

RuntimeError: It appears that you are attempting to reference SparkContext from a broadcast variable, action, or transformation. SparkContext can only be used on the driver, not in code that it run on workers. For more information, see SPARK-5063.

UDF运行在Spark的worker节点上，而spark.sql()需要依赖Driver端的SparkContext/Session，worker节点无法直接访问Driver端的上下文，因此触发该错误。

方案1：Spark原生操作（推荐，性能最优） #

通过explode拆分品牌列表为单行，再与目标表做关联，最后聚合拼接文本。这种方式完全利用Spark的分布式计算能力，规避UDF的限制。

假设目标表名为brand_text_table，结构为city, brand, date, text（存储对应城市、品牌、日期的文本内容），代码示例：

from pyspark.sql import functions as F

# 1. 拆分品牌列表为单行数据
exploded_df = original_df.withColumn("brand", F.explode(F.col("brand")))

# 2. 与目标表关联，获取对应文本
joined_df = exploded_df.join(
    F.table("brand_text_table"),
    on=["city", "brand", "date"],
    how="left"  # 根据需求选择join类型，left保留原表所有数据
)

# 3. 按原表的城市、日期、weight分组，拼接文本
result_df = joined_df.groupBy("city", "weight", "date")\
    .agg(F.concat_ws(" ", F.collect_list("text")).alias("concatenated_text"))

# 查看结果
result_df.show(truncate=False)

如果需要保留原品牌列表的顺序（Spark 2.4+支持），可用posexplode记录位置，聚合时按位置排序：

# 拆分时记录品牌的原始位置
exploded_df = original_df.select(
    "city", "weight", "date",
    F.posexplode(F.col("brand")).alias("pos", "brand")
)

# 关联后按位置排序再拼接
joined_df = exploded_df.join(F.table("brand_text_table"), ["city", "brand", "date"], "left")

result_df = joined_df.groupBy("city", "weight", "date")\
    .agg(F.concat_ws(" ", F.collect_list("text").orderBy("pos")).alias("concatenated_text"))

方案2：使用Pandas UDF（仅特殊场景使用） #

如果必须依赖UDF逻辑，可使用Pandas UDF（向量UDF），但需先将目标表数据广播到worker节点，避免在UDF中查询数据库。仅适合目标表数据量较小的场景，否则会占用过多worker内存。

代码示例：

from pyspark.sql import functions as F
from pyspark.sql.functions import pandas_udf
import pandas as pd

# 加载目标表数据并广播
brand_text_pd = spark.table("brand_text_table").toPandas()
broadcast_brand_text = spark.sparkContext.broadcast(brand_text_pd)

# 定义Pandas UDF
@pandas_udf("string")
def concat_brand_text(cities: pd.Series, brands: pd.Series, dates: pd.Series) -> pd.Series:
    text_df = broadcast_brand_text.value
    result = []
    for city, brand_list, date in zip(cities, brands, dates):
        # 筛选当前城市、日期下的品牌文本
        filtered = text_df[(text_df["city"] == city) & (text_df["date"] == date)]
        # 按原品牌列表顺序拼接文本
        texts = [
            filtered[filtered["brand"] == b]["text"].iloc[0] 
            if not filtered[filtered["brand"] == b].empty 
            else "" 
            for b in brand_list
        ]
        result.append(" ".join(texts))
    return pd.Series(result)

# 应用UDF生成结果
result_df = original_df.withColumn(
    "concatenated_text",
    concat_brand_text(F.col("city"), F.col("brand"), F.col("date"))
)

result_df.show(truncate=False)

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