如何将后定义的参数传入PySpark SQL查询

方案1：使用Spark参数化查询（推荐，避免SQL注入） #

Spark支持通过params参数传递变量，无需字符串拼接，既安全又能避免变量作用域问题：

# 定义带占位符的SQL模板
sql_template = """
select
    Country_Name as country,
    State_Name as state,
    City_Name as city,
    avg(Order_Rate)
from last_day 
where Country_Name = 'USA' and Account_ID = {account_id}
group by 1,2,3;
"""

for orders in order_list:
    account_id = orders["Account_ID"]
    # 传递参数执行查询
    spark.sql(sql_template, params={"account_id": account_id})

方案2：封装SQL生成函数 #

把SQL生成逻辑封装成独立函数，循环中调用函数生成查询语句，代码结构更清晰：

def get_account_query(account_id):
    return f"""
select
    Country_Name as country,
    State_Name as state,
    City_Name as city,
    avg(Order_Rate)
from last_day 
where Country_Name = 'USA' and Account_ID = "{account_id}"
group by 1,2,3;
"""

for orders in order_list:
    account_id = orders["Account_ID"]
    spark.sql(get_account_query(account_id))

方案3：批量查询（效率最优） #

如果不需要逐个处理单账号结果，可以一次性提取所有账号ID，用IN子句批量查询，减少Spark作业提交次数：

# 提取所有Account_ID
account_ids = [orders["Account_ID"] for orders in order_list]
# 格式化IN子句的字符串参数
account_ids_str = ",".join([f"'{id}'" for id in account_ids])

sql_query = f"""
select
    Country_Name as country,
    State_Name as state,
    City_Name as city,
    Account_ID,
    avg(Order_Rate)
from last_day 
where Country_Name = 'USA' and Account_ID in ({account_ids_str})
group by 1,2,3,4;
"""

# 一次执行查询，结果包含所有目标账号的数据
result = spark.sql(sql_query)

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