Spark 3.3下Spark BigQuery连接器参数化查询的安全方案咨询

针对你在Spark 3.3环境下无法使用Spark 3.4新增的参数化SQL支持，同时要避免SQL注入的需求，提供以下几种替代方案：

方案一：利用Spark Dataset API后置过滤 #

先读取不带过滤条件的基础数据集，再通过Spark的类型安全Dataset API进行过滤。Spark Catalyst优化器会自动将过滤条件下推到BigQuery执行，既避免SQL拼接，又保证性能：

// 读取基础数据集
Dataset<Row> baseDs = spark.read()
    .format("bigquery")
    .option("query", "SELECT col1, col2, x FROM your_table")
    .load();

// 类型安全的参数化过滤
int cutoff = 100;
Dataset<Row> filteredDs = baseDs.filter(baseDs.col("x").gt(cutoff));

方案二：借助BigQuery Java客户端原生参数化查询 #

直接使用BigQuery官方Java客户端构建参数化查询，执行后将结果转换为Spark Dataset，完全利用BigQuery的原生参数化支持杜绝注入：

import com.google.cloud.bigquery.*;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.types.StructType;
import java.util.stream.Collectors;

// 初始化BigQuery客户端
BigQuery bigquery = BigQueryOptions.getDefaultInstance().getService();

// 构建带命名参数的查询配置
String sql = "SELECT col1, col2, x FROM your_table WHERE x > @cutoff";
QueryJobConfiguration queryConfig = QueryJobConfiguration.newBuilder(sql)
    .addNamedParameter("cutoff", QueryParameterValue.int64(100))
    .build();

// 执行查询并转换为Spark Dataset
TableResult result = bigquery.query(queryConfig);
StructType schema = ...; // 可从TableResult的Schema手动映射或自动推断
Dataset<Row> ds = spark.createDataFrame(
    result.iterateAll().stream()
        .map(row -> RowFactory.create(row.getValues()))
        .collect(Collectors.toList()),
    schema
);

方案三：自定义安全参数替换工具 #

如果必须通过Spark BigQuery连接器的query参数传递SQL，可实现一个简单的参数替换工具，针对不同类型做严格转义，降低注入风险：

import org.apache.commons.lang3.StringEscapeUtils;

public class SafeSqlParamReplacer {
    public static String replaceNamedParam(String sqlTemplate, String paramName, Object value) {
        String escapedVal;
        if (value instanceof String) {
            escapedVal = "'" + StringEscapeUtils.escapeSql((String) value) + "'";
        } else if (value instanceof Number) {
            escapedVal = value.toString();
        } else if (value instanceof Boolean) {
            escapedVal = ((Boolean) value) ? "TRUE" : "FALSE";
        } else {
            throw new IllegalArgumentException("不支持的参数类型");
        }
        return sqlTemplate.replace("@" + paramName, escapedVal);
    }
}

// 使用示例
String sqlTemplate = "SELECT col1, col2, x FROM your_table WHERE x > @cutoff";
String safeSql = SafeSqlParamReplacer.replaceNamedParam(sqlTemplate, "cutoff", 100);
Dataset<Row> ds = spark.read().format("bigquery").option("query", safeSql).load();

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