#java #apache-spark #kotlin #delta-lake
#java #apache-spark #kotlin #delta-lake
Вопрос:
Моя цель — постоянно помещать входящие файлы parquet в 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();;
Как я могу получить отфильтрованные данные из delta Lake?
Ответ №1:
Вы пробовали использовать foreachBatch?
Это позволяет использовать все пакетные функции для потоковой передачи, и вы также можете несколько контролировать количество файлов, которые вы записываете в delta lake.
Комментарии:
1. используйте «git fetch», чтобы получить всю удаленную ветку в вашу локальную.