问题描述
我的目标是将传入的镶木地板文件连续放入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();;
如何从三角洲湖中获取过滤后的数据?
解决方法
您尝试使用foreachBatch吗?
它将所有类似批处理的功能引入流中,并且您还可以稍微控制要写入delta lake的文件数量。