如何用Python为PySpark添加自定义流式Oracle数据源支持？

前置说明 #

• spark.read.format('...')（批量读取）和spark.readStream.format('...')（流式读取）是两种不同操作，对应批量读取和结构化流两个完全独立的概念，本次需求聚焦于流式读取。

需求概述 #

我需要从本地Oracle数据库的多关联表中读取数据并转换为Spark流，计划实现一个DataSourceV2类型的自定义数据源，用来从Oracle读取数据并维护检查点信息，以支持流读取的调度与恢复，最终希望写出如下简洁代码：

streaming_oracle_df = spark.readStream \
    .format("custom_oracle") \
    .option("oracle_jdbc_str", "jdbc:...") \
    .option("custom_option1", 123) \
    .load()

streaming_oracle_df.writeStream \
    .trigger(availableNow=True) \
    .format('delta') \
    .option('checkpointLocation', 's3://bucket/checkpoint/dim_customer') \
    .start('s3://bucket/tables/dim_customer')

疑问 #

是否可以用Python实现该功能？还是必须使用Java？Scala是否也可行？

语言可行性分析 #

1. Java/Scala：完全可行，为首选方案
   Spark的DataSourceV2 API基于JVM设计，Java和Scala作为Spark原生开发语言，可完整实现自定义流式数据源。通过实现ReadSupport、StreamingReadSupport等核心接口，能处理Oracle数据的分片读取、检查点状态维护、增量数据拉取等全部逻辑，完全匹配需求。

2. Python：无法直接实现DataSourceV2流式数据源
   PySpark并未暴露DataSourceV2的底层接口，无法直接用Python编写自定义流式数据源的核心逻辑。若要在PySpark中使用自定义Oracle流式数据源，需先通过Java或Scala完成数据源的开发，再在PySpark代码中通过.format("custom_oracle")调用该JVM端实现。

Python替代实现方案 #

若倾向用Python完成整体逻辑，可通过结构化流foreachBatch机制模拟实现类似效果：

• 在foreachBatch的批次处理逻辑中，用JDBC批量读取Oracle的增量数据（以时间戳、自增ID等作为偏移量标记）
• 手动将偏移量信息写入检查点存储（如Delta表、外部KV存储）
• 流启动时从检查点读取上次的偏移量，实现增量拉取
  该方式虽能达成目标，但代码复杂度高于自定义DataSourceV2，且性能略逊于原生流式数据源。

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