Databricks中基于组合表匹配行实现客户分类的方案问询

需要通过Clients表与Combinations表的行匹配对客户分类：若Clients表的行与Combinations表的行完全匹配，则客户归为B类，否则归为A类。当前用嵌套循环实现逻辑，但在Databricks中运行报错，现有代码如下：

for i,j in select_df.iterrows():
      for u,v in dfCombinacionesDias.iterrows():
          if (
              (select_df["MONDAY"][i] == registro["LUNES"][u]) 
              and (select_df["TUESDAY"][i] == registro["MARTES"][u]) 
              and (select_df["WEDNESDAY"][i] == registro["MIERCOLES"][u]) 
              and (select_df["THURSDAY"][i] == registro["JUEVES"][u]) 
              and (select_df["FRIDAY"][i] == registro["VIERNES"][u]) 
              and (select_df["SATURDAY"][i] == registro["SABADO"][u]) 
              and (select_df["SUNDAY"][i] == registro["DOMINGO"][u])
          ):
              Sector = "B"
          else:
              Sector = "A"
        
vSubSeq = "('{}','{}')".format(select_df["IDClient"][i],Sector)
sqlInsertSequence = "Insert into {0}.{1} values {2}".format(dSCHEMA, Table, vSubSeq,vdataDeltaPath)
print(sqlInsertSequence)
dfTables = spark.sql(sqlInsertSequence)

表结构说明：

• Clients表（对应代码中的select_df）：包含IDClient、MONDAY、TUESDAY、WEDNESDAY、THURSDAY、FRIDAY、SATURDAY、SUNDAY字段
• Combinations表（对应代码中的dfCombinacionesDias）：包含LUNES、MARTES、MIERCOLES、JUEVES、VIERNES、SABADO、DOMINGO字段（与Clients表的星期字段一一对应，仅语言命名不同）

原代码核心问题 #

1. 违背Spark分布式设计：iterrows()是Pandas逐行遍历方法，Spark DataFrame是分布式数据集，逐行循环会彻底丧失Spark的并行计算优势，还易引发内存溢出、性能暴跌等问题。
2. 逻辑错误：嵌套循环中每一次内层迭代都会覆盖Sector值，最终仅保留最后一次循环的判断结果，无法正确识别客户是否与任意一行Combinations匹配。
3. SQL注入风险：通过字符串拼接生成INSERT语句存在安全隐患，且Spark原生提供更安全的写入方式。

最优实现方案（Spark原生API/SQL） #

利用Spark的左连接+条件判断实现，完全适配分布式计算场景：

方法1：Spark DataFrame API实现 #

# 重命名Combinations表的星期字段，与Clients表统一，方便关联
combinations_renamed = dfCombinacionesDias.withColumnRenamed("LUNES", "MONDAY")\
    .withColumnRenamed("MARTES", "TUESDAY")\
    .withColumnRenamed("MIERCOLES", "WEDNESDAY")\
    .withColumnRenamed("JUEVES", "THURSDAY")\
    .withColumnRenamed("VIERNES", "FRIDAY")\
    .withColumnRenamed("SABADO", "SATURDAY")\
    .withColumnRenamed("DOMINGO", "SUNDAY")

# 左连接后判断是否匹配，生成分类结果
client_classified = select_df.join(combinations_renamed, 
                                  on=["MONDAY", "TUESDAY", "WEDNESDAY", "THURSDAY", "FRIDAY", "SATURDAY", "SUNDAY"],
                                  how="left")\
    .withColumn("Sector", 
               when(col("MONDAY").isNotNull(), "B")  # 左连接匹配成功则字段非空，标记为B类
               .otherwise("A"))\
    .select("IDClient", "Sector")

# 写入目标表（Spark原生写入，避免SQL拼接风险）
client_classified.write.mode("append").saveAsTable(f"{dSCHEMA}.{Table}")

方法2：Spark SQL实现 #

先将表注册为临时视图，再执行SQL语句：

# 注册临时视图
select_df.createOrReplaceTempView("clients")
dfCombinacionesDias.createOrReplaceTempView("combinations")

# 执行SQL完成分类
classified_sql = f"""
SELECT 
    c.IDClient,
    CASE WHEN comb.LUNES IS NOT NULL THEN 'B' ELSE 'A' END AS Sector
FROM clients c
LEFT JOIN combinations comb
    ON c.MONDAY = comb.LUNES
    AND c.TUESDAY = comb.MARTES
    AND c.WEDNESDAY = comb.MIERCOLES
    AND c.THURSDAY = comb.JUEVES
    AND c.FRIDAY = comb.VIERNES
    AND c.SATURDAY = comb.SABADO
    AND c.SUNDAY = comb.DOMINGO
"""

client_classified = spark.sql(classified_sql)

# 写入目标表
client_classified.write.mode("append").saveAsTable(f"{dSCHEMA}.{Table}")

方案优势 #

• 完全利用Spark分布式计算能力，处理大规模数据时性能远超逐行循环
• 逻辑精准：左连接后只要匹配到任意一行Combinations，就标记为B类，否则为A类
• 避免SQL注入风险，代码简洁易维护

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