Merge pull request #27 from TeoMeWhy/fix

Fix

src/gold/ingestao.py

@@ -0,0 +1,33 @@
+# Databricks notebook source
+# DBTITLE 1,IMPORTS
+import tqdm
+import sys
+import datetime
+
+sys.path.insert(0, "../lib")
+
+import utils
+import ingestors
+
+# COMMAND ----------
+
+# DBTITLE 1,SETUP
+catalog = "gold"
+schemaname = 'upsell'
+tablename = dbutils.widgets.get("tablename")
+
+start = dbutils.widgets.get("dt_start") # now
+stop = dbutils.widgets.get("dt_stop") # now
+
+if start == datetime.datetime.now().strftime('%Y-%m-%d'):
+    start = (datetime.datetime.now() - datetime.timedelta(days=1)).strftime("%Y-%m-%d")
+
+
+# COMMAND ----------
+
+ingestor = ingestors.IngestorCubo(spark=spark,
+                                  catalog=catalog,
+                                  schemaname=schemaname,
+                                  tablename=tablename)
+
+ingestor.backfill(start, stop)

src/lib/ingestors.py

@@ -163,13 +163,19 @@ def load(self, **kwargs):
         return df
 
     def save(self, df, dt_ref):
+        self.spark.sql(f"DELETE FROM {self.table} WHERE dtRef = '{dt_ref}'")
+
         (df.write
-           .mode("overwrite")
-           .option("replaceWhere", f"dtRef = '{dt_ref}'")
+           .mode("append")
            .saveAsTable(self.table))
 
     def backfill(self, dt_start, dt_stop):
         dates = utils.date_range(dt_start, dt_stop)
+
+        if not utils.table_exists(self.spark, self.catalog, self.schemaname, self.tablename):
+            df = self.load(dt_ref=dates.pop(0))
+            df.write.saveAsTable(self.table)
+
         for dt in tqdm.tqdm(dates):
             df = self.load(dt_ref=dt)
             self.save(df=df, dt_ref=dt)