参数化SQL查询扩展方法及PySpark查询添加GROUP BY方案

参数化查询的扩展可以从这些方向实现：

• 动态添加查询条件：通过传入参数列表动态拼接WHERE子句，全程使用参数绑定而非字符串拼接，彻底避免SQL注入。比如在Python中用%s或?作为占位符，将参数传入执行方法。
• 动态列选择：允许传入需要查询的列名列表，拼接进SELECT子句，同时要对列名做合法性校验（比如匹配预设白名单），防止恶意注入。
• 动态排序与分页：接收排序字段、排序方向、页码、每页条数等参数，动态生成ORDER BY和LIMIT/OFFSET子句，排序字段需通过白名单校验，避免非法字段。
• 动态多表关联：根据参数决定是否关联其他表，或调整关联条件，关联值同样用参数绑定处理。
• 参数化聚合逻辑：支持动态选择聚合函数（COUNT/SUM等）或分组字段，聚合函数和分组字段必须做白名单校验，确保是合法的SQL元素。

下面提供两种修改方案，分别对应固定分组和灵活分组的场景：

方案1：固定按'xyz'分组 #

直接修改函数内的SQL语句，加入GROUP BY子句并保留分组字段：

from pyspark import SparkContext, SparkConf
from pyspark.sql import HiveContext
from pyspark.sql import SQLContext
from pyspark.sql import SparkSession
from pyspark.sql.types import *

db = 'database'
schema = 'Schema'

def getCountByXyz(table):
    # 调整SQL，选择xyz字段和统计数，按xyz分组
    string = f"select xyz, count(*) as ct from {db}.{schema}.{table} group by xyz"
    df = spark.read.format(snowflake_name)\
                         .options(**sfOptions)\
                         .option('query', string).load()
    return df

方案2：支持动态分组字段（更灵活） #

如果后续需要切换分组字段，可以把分组字段作为参数传入，同时添加白名单校验避免SQL注入：

from pyspark import SparkContext, SparkConf
from pyspark.sql import HiveContext
from pyspark.sql import SQLContext
from pyspark.sql import SparkSession
from pyspark.sql.types import *

db = 'database'
schema = 'Schema'

def getCountByGroup(table, group_col='xyz'):
    # 白名单校验，仅允许合法的分组字段
    allowed_columns = ['xyz', 'col_a', 'col_b']  # 根据实际表字段维护
    if group_col not in allowed_columns:
        raise ValueError(f"非法分组字段：{group_col}")
    
    string = f"select {group_col}, count(*) as ct from {db}.{schema}.{table} group by {group_col}"
    df = spark.read.format(snowflake_name)\
                         .options(**sfOptions)\
                         .option('query', string).load()
    return df

说明：

• 固定分组方案适合只需要按'xyz'统计的场景，代码简洁直接。
• 动态分组方案提升了灵活性，白名单校验能有效防止SQL注入，推荐在需要多场景分组时使用。

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