PySpark通过Databricks JDBC分区读取数据返回0条记录的问题排查

核心问题分析 #

你遇到的src_df.count()=0主要是两个关键错误导致：

1. dbtable参数误用：当使用partitionColumn等分区配置时，Spark会将dbtable的值视为表/视图名称，而非直接执行的SQL子查询。你传入的(select ...) as q1会被当成表名处理，导致Spark生成的最终SQL语法错误，返回空结果。而无分区时用query参数可以正常运行，是因为query会直接执行传入的SQL语句。
2. upperBound取值错误：你的数据集有200万条记录，但upperBound设为60000，即使语法问题解决，Spark也只会查询row_number <=60000的范围，无法覆盖全部数据。

排查步骤 #

• 查看Spark执行的真实SQL：开启JDBC debug日志（设置log4j.logger.org.apache.spark.sql.execution.datasources.jdbc=DEBUG），检查发送到Databricks的SQL语句，确认是否存在语法错误。
• 验证分片查询有效性：手动执行Spark生成的分片查询（例如SELECT * FROM (你的完整子查询) WHERE row_number BETWEEN 1 AND 250000），在Databricks中运行，确认是否有数据返回。
• 确认row_number的实际范围：在Databricks中执行原SQL，查看max(row_number)的值，确保upperBound与实际最大值匹配。

解决方案 #

方案1：使用query参数配合分区占位符 #

当需要通过query参数执行子查询并分区时，需在SQL中预留Spark的分区过滤占位符，让Spark自动替换为对应范围条件：

1. 调整SQL查询（补充真实表名，添加占位符）：

sqlquery= "select row_number() over(order by (select null)) as row_number, artifact1_name, artifact2_name from your_actual_table where ${partitionColumn} between ${lowerBound} and ${upperBound}"

2. 修改PySpark读取代码：

src_df= spark.read.format("jdbc"). \
        option("url", url). \
        option("query", sqlquery). \
        option("fetchSize", 20000). \
        option("partitionColumn", "row_number"). \
        option("lowerBound", 1). \
        option("upperBound", 2000000).  # 匹配实际数据的最大row_number
        option("numPartitions", 8). \
        option("driver", driver). \
        option("user", user). \
        option("password", PAT_token).load()

方案2：创建临时视图后用dbtable读取 #

先在Databricks中创建包含row_number的临时视图，再通过dbtable指向视图进行分区读取：

1. 创建临时视图：

spark.read.format("jdbc"). \
        option("url", url). \
        option("query", "create or replace temporary view temp_view as select row_number() over(order by (select null)) as row_number, artifact1_name, artifact2_name from your_actual_table"). \
        option("driver", driver). \
        option("user", user). \
        option("password", PAT_token).load()

2. 分区读取视图：

src_df= spark.read.format("jdbc"). \
        option("url", url). \
        option("dbtable", "temp_view"). \
        option("fetchSize", 20000). \
        option("partitionColumn", "row_number"). \
        option("lowerBound", 1). \
        option("upperBound", 2000000). \
        option("numPartitions", 8). \
        option("driver", driver). \
        option("user", user). \
        option("password", PAT_token).load()

后续处理 #

读取完成后，仍可通过final_df=src_df.drop("row_number")删除分区列，再写入PostgreSQL即可。

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