Hi,
I have the following problem: on a medallion architecture on a bronze volume I get files every month containing the data for each sensor reading during the period 1 of month 00:00 to last day 23:00. I have a manual job that calls the python files to load and transform from landing -> bronze-> silver. Now I want to add the aggregated data to a gold table. I have the following code:
df_silver_swell_metrics = (
spark.readStream
.format("delta")
.table(f"cor_project.silver.swell_metrics")
.withWatermark("datetime", "0 second")
)
df_silver_swell_metrics_transformed = (
df_silver_swell_metrics
.groupBy(
F.window("datetime", "1 day").alias("window"),
"coast_name"
).agg(
...
)
)
df_gold_wave_daily_summary = (
df_silver_swell_metrics_transformed
.select(
...
)
)
(
df_gold_wave_daily_summary.writeStream
.format("delta")
.outputMode("append")
.trigger(availableNow=True)
.option("checkpointLocation", f"/Volumes/cor_{ambiente}/gold/data/checkpoints/wave_daily_summary")
.toTable(f"cor_project.gold.wave_daily_summary")
)
This works it doesn't repeat data that has already been processed but it always misses the last day of the month. I have asked in other groups and they told me to use batch processing. Any thoughts on posible solutions?
Thanks for any comment
