PySpark操作Kudu流遇读取不支持错误，咨询读取及写入方案

希望使用PySpark从Kudu表流式读取数据，处理后流式写入另一个Kudu表，尝试的驱动代码如下：

from pyspark.sql import SparkSession

spark = SparkSession \
    .builder \
    .appName("Kudu-stream") \
    .getOrCreate()

df = spark \
    .readStream \
    .format("org.apache.kudu.spark.kudu") \
    .option("kudu.master","example:7051") \
    .option("kudu.table","db.source_test") \
    .load()

def saveToKudu(batchDF, batchID):
  batchDF.write \
  .format("org.apache.kudu.spark.kudu") \
  .option("kudu.master","example:7051") \
  .option("kudu.table","db.sink_test") \
  .mode("append") \
  .save()

query = df \
  .writeStream \
  .outputMode("append") \
  .foreachBatch(saveToKudu) \
  .option("checkpointLocation","hdfs://example:8020/path/to/checkpoint/") \
  .start()

query.awaitTermination()

运行后报错：

py4j.protocol.Py4JJavaError: An error occurred while calling o40.load.
: java.lang.UnsupportedOperationException: Data source org.apache.kudu.spark.kudu does not support streamed reading

疑问：

1. 是否存在直接从Kudu表执行readStream的方法？
2. 能否不使用foreach或foreachBatch实现writeStream到Kudu表？

关于Kudu的流式读取（readStream） #

Kudu官方Spark连接器不支持直接流式读取，这是你报错的核心原因。要实现类流式读取Kudu的效果，可采用两种方案：

• 基于时间戳/递增ID的微批轮询：定期读取Kudu表中新增数据（比如通过update_time或自增ID过滤上次轮询后的记录），模拟流式处理逻辑。示例代码如下：

from pyspark.sql import SparkSession
from pyspark.sql.functions import current_timestamp
import time

spark = SparkSession.builder.appName("Kudu-polling").getOrCreate()

last_poll_time = None
poll_interval = 60  # 每60秒轮询一次

while True:
    # 根据上次轮询时间过滤数据
    if last_poll_time:
        df = spark.read.format("org.apache.kudu.spark.kudu")\
            .option("kudu.master", "example:7051")\
            .option("kudu.table", "db.source_test")\
            .load()\
            .filter(f"update_time > '{last_poll_time}'")
    else:
        df = spark.read.format("org.apache.kudu.spark.kudu")\
            .option("kudu.master", "example:7051")\
            .option("kudu.table", "db.source_test")\
            .load()
    
    # 替换为你的数据处理逻辑
    processed_df = df
    
    # 写入目标Kudu表
    processed_df.write.format("org.apache.kudu.spark.kudu")\
        .option("kudu.master", "example:7051")\
        .option("kudu.table", "db.sink_test")\
        .mode("append")\
        .save()
    
    # 更新上次轮询时间
    last_poll_time = current_timestamp().cast("string").collect()[0][0]
    time.sleep(poll_interval)

• 借助Kafka中间层：如果业务允许，通过CDC工具（如Debezium）将Kudu的变更数据同步到Kafka，再用Spark从Kafka进行流式读取，实现真正的实时处理。

关于Kudu的流式写入（writeStream） #

Kudu官方Spark连接器不支持直接writeStream写入，必须通过foreachBatch或foreach实现批量写入。其中foreachBatch是推荐方案，它能利用Spark批量写入优化，效率远高于单条写入的foreach。

你当前代码中foreachBatch的写法是正确的，只需替换流式读取部分为上述轮询或Kafka CDC方案即可正常运行。

内容的提问来源于stack exchange，提问作者Luka Klinčić