fix: remove empty tables

basedosdados · Sep 26, 2023 · 79f4259 · 79f4259
1 parent 64ae7ee
commit 79f4259
Show file tree

Hide file tree

Showing 2 changed files with 29 additions and 19 deletions.
diff --git a/pipelines/datasets/br_cgu_servidores_executivo_federal/flows.py b/pipelines/datasets/br_cgu_servidores_executivo_federal/flows.py
@@ -94,13 +94,14 @@
     #     wait=outputs_path_by_table,
     # )
 
-    create_table_and_upload_to_gcs(
-        data_path=outputs_path_by_table["servidores_cadastro"],
-        dataset_id=dataset_id,
-        table_id="servidores_cadastro",
-        dump_mode="append",
-        wait=outputs_path_by_table,
-    )
+    if "servidores_cadastro" in outputs_path_by_table:
+        create_table_and_upload_to_gcs(
+            data_path=outputs_path_by_table["servidores_cadastro"],
+            dataset_id=dataset_id,
+            table_id="servidores_cadastro",
+            dump_mode="append",
+            wait=outputs_path_by_table,
+        )
 
     # create_table_and_upload_to_gcs(
     #     data_path=outputs_path_by_table["reserva_reforma_militares_cadastro"],

diff --git a/pipelines/datasets/br_cgu_servidores_executivo_federal/tasks.py b/pipelines/datasets/br_cgu_servidores_executivo_federal/tasks.py
@@ -94,15 +94,24 @@ def make_partitions(tables: list[tuple[str, pd.DataFrame]]) -> dict[str, str]:
         os.mkdir(output)
 
     for table_name, df in tables:
-        savepath = f"{output}/{table_name}"
-
-        if not os.path.exists(savepath):
-            os.mkdir(savepath)
-
-        to_partitions(
-            data=df,
-            partition_columns=["ano", "mes"],
-            savepath=savepath,
-        )
-
-    return {table_name: f"{output}/{table_name}" for table_name, _ in tables}
+        if len(df) > 0:
+            savepath = f"{output}/{table_name}"
+
+            if not os.path.exists(savepath):
+                os.mkdir(savepath)
+
+            log(f"{table_name=}")
+            log(f"{df.columns=}")
+            log(df.head())
+
+            to_partitions(
+                data=df,
+                partition_columns=["ano", "mes"],
+                savepath=savepath,
+            )
+        else:
+            log(f"{table_name=} is empty")
+
+    return {
+        table_name: f"{output}/{table_name}" for table_name, df in tables if len(df) > 0
+    }