Commit: 2aeecd2bac13f8a430891b9f42f01bedcc43d02c - myprojects-hg-reborn (hg)

Commit MetaInfo

• Revisão: 2aeecd2bac13f8a430891b9f42f01bedcc43d02c (tree)
• Hora: 2024-09-18 04:08:05
• Autor: Lorenzo Isella <lorenzo.isella@gmai...>
• Commiter: Lorenzo Isella

Mensagem de Log

A cleanup of the code to merge the various parquet files.

Mudança Sumário

• modified: R-codes/merge_tam_parquet.R

Diff

diff -r 69e147994e03 -r 2aeecd2bac13 R-codes/merge_tam_parquet.R
--- a/R-codes/merge_tam_parquet.R	Tue Sep 17 20:11:43 2024 +0200
+++ b/R-codes/merge_tam_parquet.R	Tue Sep 17 21:08:05 2024 +0200
@@ -11,16 +11,16 @@
 
 add_tctf <- 1
 
-read_all <- 0
+## read_all <- 1
 
 
 
-tctf <- open_dataset("TCTF_cases.csv", format="csv")
+tctf <- open_dataset("../input/csv_files/TCTF_cases.csv", format="csv")
 
 
-if (read_all==1){
+## if (read_all==1){
 
-df <- open_dataset("./input_all/", unify_schemas = TRUE)  |>
+df <- open_dataset("../input/parquet-files/", unify_schemas = TRUE)  |>
     select(everything())   ### I need to select the columns in any case.
 
 
@@ -32,10 +32,13 @@
 
 }
 
+
+remove_files_with_pattern("../output/*parquet")
+
 write_dataset(
   df,
   format = "parquet",
-  path = "./data_output_all/",
+  path = "../output/",
   basename_template="tam_finale_20-03-2024-{i}.parquet" ,
   max_rows_per_file = 5e5
 )
@@ -48,82 +51,89 @@
 
     print("I start writing the csv files")
 
+remove_files_with_pattern("../output/flat_files/*csv")
+    
 write_dataset(
   df,
   format = "csv",
-  path = "./data_output_all/flat_files/",
+  path = "../output/flat_files/",
   basename_template="tam_finale_20-03-2024-{i}.csv" ,
   max_rows_per_file = 5e5,
   max_open_files=5)
 
+system("csvstack ../output/flat_files/*csv > ../output/flat_files/tam_complete.csv")
 
-system("gzip ./data_output_all/flat_files/*csv")
 
+    
+system("gzip -f -9 ../output/flat_files/tam_complete.csv")
+
+remove_files_with_pattern("../output/flat_files/*csv")
+    
 
 }
 
 
 
     
-} else{
-
-df <- open_dataset("./input_all/extra", unify_schemas = TRUE)  |>
-    select(everything())   ### I need to select the columns in any case.
-
-if (add_tctf==1){
-    
-    df <- df |>
-        left_join(y=tctf, by=c("case_reference"="sa_case_number")) |>
-        mutate(is_tctf_case=if_else(is.na(is_tctf_case), "No", is_tctf_case))
-}
+## } else{
 
-write_dataset(
-  df,
-  format = "parquet",
-  path = "./data_output_all/TAM_ES_RO_SI/parquet/",
-  basename_template="tam_tctf_20-03-2024-{i}.parquet" ,
-  max_rows_per_file = 5e5
-)
+## df <- open_dataset("../input_all/extra", unify_schemas = TRUE)  |>
+##     select(everything())   ### I need to select the columns in any case.
 
-print("writing the TAM compressed csv file")
+## if (add_tctf==1){
     
-write_csv_arrow(
-  df,"./data_output_all/TAM_ES_RO_SI/flat_file/transparency_tctf-20-03-2024.csv.gz")
+##     df <- df |>
+##         left_join(y=tctf, by=c("case_reference"="sa_case_number")) |>
+##         mutate(is_tctf_case=if_else(is.na(is_tctf_case), "No", is_tctf_case))
+## }
+
+## write_dataset(
+##   df,
+##   format = "parquet",
+##   path = "../data_output_all/TAM_ES_RO_SI/parquet/",
+##   basename_template="tam_tctf_20-03-2024-{i}.parquet" ,
+##   max_rows_per_file = 5e5
+## )
+
+## print("writing the TAM compressed csv file")
+    
+## write_csv_arrow(
+##   df,"../data_output_all/TAM_ES_RO_SI/flat_file/transparency_tctf-20-03-2024.csv.gz")
 
 
 
     
-### now we do Poland
+## ### now we do Poland
 
-df<- open_dataset("./input_all/poland/", unify_schemas = TRUE)  |>
-    select(everything())   ### I need to select the columns in any case.
+## df<- open_dataset("../input_all/poland/", unify_schemas = TRUE)  |>
+##     select(everything())   ### I need to select the columns in any case.
 
 
-    if (add_tctf==1) {
-
-        df <- df |>
-        left_join(y=tctf, by=c("case_reference"="sa_case_number")) |>
-        mutate(is_tctf_case=if_else(is.na(is_tctf_case), "No", is_tctf_case))
-
-}
+##     if (add_tctf==1) {
 
-write_dataset(
-  df,
-  format = "parquet",
-  path = "./data_output_all/poland/arrow/",
-  basename_template="poland_tctf_20-03-2024-{i}.parquet" ,
-  max_rows_per_file = 5e5
-)
+##         df <- df |>
+##         left_join(y=tctf, by=c("case_reference"="sa_case_number")) |>
+##         mutate(is_tctf_case=if_else(is.na(is_tctf_case), "No", is_tctf_case))
 
-print("writing Poland compressed csv file")
+## }
 
-write_csv_arrow(
-  df,"./data_output_all/poland/flat_file/poland_tctf-20-03-2024.csv.gz")
+## write_dataset(
+##   df,
+##   format = "parquet",
+##   path = "../data_output_all/poland/arrow/",
+##   basename_template="poland_tctf_20-03-2024-{i}.parquet" ,
+##   max_rows_per_file = 5e5
+## )
+
+## print("writing Poland compressed csv file")
+
+## write_csv_arrow(
+##   df,"../data_output_all/poland/flat_file/poland_tctf-20-03-2024.csv.gz")
     
 
     
     
-}
+## }