Databricks通过Redshift JDBC读数据触发参数冲突异常

使用闲置30分钟自动关闭的Databricks集群（13.3 LTS，含Apache Spark 3.4.1、Scala 2.12），实现从Redshift读取表数据并写入Snowflake，代码如下：

df = spark.read \
  .format("redshift") \
  .option("url", jdbc_url) \
  .option("user", user) \
  .option("password", password) \
  .option("dbtable", "schem_name.table_name") \
  .option("partitionColumn", "date_col1")\
  .option("lowerBound", "2023-11-05")\
  .option("upperBound", "2024-03-23")\
  .option("numPartitions", "500")\
  .load()\
  .filter("date_col1>dateadd(month ,-6,current_date)")\
  .filter(col("country_col").isin('India', 'China', 'Japan', 'Germany', 'United Kingdom', 'Brazil', 'United States', 'Canada'))
df1 = df.repartition(900)#Data is skedwed for that partition column, so repartitioning to 1* num cores in cluster for even dist
df1.write.format("snowflake") \
        .option("host", host_sfl) \
        .option("user", user_sfl) \
        .option('role', role_sfl) \
        .option("password", password_sfl) \
        .option("database", database_sfl) \
        .option("sfWarehouse", warehouse_sfl) \
        .option("schema",'schema_name')\
        .option("dbtable",'target_table_name')\
        .mode('Overwrite') \
        .save()

运行时抛出异常（未使用query参数）：

IllegalArgumentException: requirement failed:
Options 'query' and 'partitionColumn' can not be specified together.
Please define the query using dbtable option instead and make sure to qualify
the partition columns using the supplied subquery alias to resolve any ambiguity.
Example :
spark.read.format("jdbc")
.option("url", jdbcUrl)
.option("dbtable", "(select c1, c2 from t1) as subq")
.option("partitionColumn", "c1")
.option("lowerBound", "1")
.option("upperBound", "100")
.option("numPartitions", "3")
.load()

关联现象 #

• 注释重分区及Snowflake写入代码，仅执行count()时结果正确；
• 执行count()后将.format("redshift")改为JDBC格式，代码可正常运行；
• 集群重启后首次运行任务必失败，需手动执行count再改JDBC才能正常运行。

1. Redshift数据源的隐式转换冲突：Databricks的Redshift数据源在处理后续filter转换时，会自动将dbtable的表读取逻辑转换为query参数形式的子查询，但未清除已配置的partitionColumn等分区参数，触发JDBC底层的参数冲突校验。
2. 集群初始化状态异常：集群闲置重启后，Redshift数据源的初始化元数据或执行计划缓存存在异常，无法兼容分区参数与后续转换的组合逻辑；执行count()后Spark执行计划被触发更新，临时规避了冲突，但仅切换为原生JDBC格式才能彻底解决，说明Redshift数据源的封装逻辑存在兼容性bug。

方案1：将过滤逻辑嵌入dbtable子查询（推荐） #

把filter条件直接写入dbtable的子查询中，避免Spark将后续转换隐式转为query参数，同时保留分区读取能力：

df = spark.read \
  .format("redshift") \
  .option("url", jdbc_url) \
  .option("user", user) \
  .option("password", password) \
  .option("dbtable", "(SELECT * FROM schem_name.table_name WHERE date_col1 > dateadd(month, -6, current_date) AND country_col IN ('India', 'China', 'Japan', 'Germany', 'United Kingdom', 'Brazil', 'United States', 'Canada')) AS subq") \
  .option("partitionColumn", "date_col1")\
  .option("lowerBound", "2023-11-05")\
  .option("upperBound", "2024-03-23")\
  .option("numPartitions", "500")\
  .load()

df1 = df.repartition(900)
df1.write.format("snowflake") \
        .option("host", host_sfl) \
        .option("user", user_sfl) \
        .option('role', role_sfl) \
        .option("password", password_sfl) \
        .option("database", database_sfl) \
        .option("sfWarehouse", warehouse_sfl) \
        .option("schema",'schema_name')\
        .option("dbtable",'target_table_name')\
        .mode('Overwrite') \
        .save()

方案2：直接使用JDBC格式读取Redshift #

既然切换为JDBC格式可正常运行，直接替换format("redshift")为format("jdbc")，并补充Redshift JDBC驱动参数（若集群未默认配置）：

df = spark.read \
  .format("jdbc") \
  .option("url", jdbc_url) \
  .option("user", user) \
  .option("password", password) \
  .option("dbtable", "schem_name.table_name") \
  .option("partitionColumn", "date_col1")\
  .option("lowerBound", "2023-11-05")\
  .option("upperBound", "2024-03-23")\
  .option("numPartitions", "500")\
  .option("driver", "com.amazon.redshift.jdbc42.Driver")\
  .load()\
  .filter("date_col1>dateadd(month ,-6,current_date)")\
  .filter(col("country_col").isin('India', 'China', 'Japan', 'Germany', 'United Kingdom', 'Brazil', 'United States', 'Canada'))

# 后续写入Snowflake代码不变
df1 = df.repartition(900)
df1.write.format("snowflake") \
        .option("host", host_sfl) \
        .option("user", user_sfl) \
        .option('role', role_sfl) \
        .option("password", password_sfl) \
        .option("database", database_sfl) \
        .option("sfWarehouse", warehouse_sfl) \
        .option("schema",'schema_name')\
        .option("dbtable",'target_table_name')\
        .mode('Overwrite') \
        .save()

方案3：缓存DataFrame规避参数重写 #

读取Redshift数据后立即执行cache()固化DataFrame，避免后续转换触发数据源的参数逻辑重写：

df = spark.read \
  .format("redshift") \
  .option("url", jdbc_url) \
  .option("user", user) \
  .option("password", password) \
  .option("dbtable", "schem_name.table_name") \
  .option("partitionColumn", "date_col1")\
  .option("lowerBound", "2023-11-05")\
  .option("upperBound", "2024-03-23")\
  .option("numPartitions", "500")\
  .load()\
  .cache()  # 缓存数据，阻断数据源参数重写逻辑

df = df.filter("date_col1>dateadd(month ,-6,current_date)")\
       .filter(col("country_col").isin('India', 'China', 'Japan', 'Germany', 'United Kingdom', 'Brazil', 'United States', 'Canada'))

df1 = df.repartition(900)
# 后续写入代码不变

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