将数据流式传输到德尔塔湖，读取过滤结果

Question

我的目标是不断地将传入的拼图文件放入delta-lake，进行查询，并将结果放入Rest API中。所有文件都在s3存储桶中。

//listen for changes
val df = spark.readStream().parquet("s3a://myBucket/folder")

//write changes to delta lake
df.writeStream()
    .format("delta")
    .option("checkpointLocation", "s3a://myBucket-processed/checkpoint")
    .start("s3a://myBucket-processed/")
    .awaitTermination() //this call lives in another thread (because it's blocking)

//this is a bad example
val query = df.select(convertedColumnNames) 
query.show()

//another bad example:
spark.readStream().format("delta").load("s3a://myBucket-processed/").select(convertedColumnNames).show()

//org.apache.spark.sql.AnalysisException: Queries with streaming sources must be executed with writeStream.start();;

我怎样才能从德尔塔湖得到过滤后的数据？

Answer 1

您是否尝试过使用foreachBatch

它为流媒体带来了所有的批处理功能，你也可以在一定程度上控制你正在写入delta的文件的数量。