iceberg partitioning (#29)

* iceberg partitioning

* table name fix

Author: Kumbus

infrastructure/modules/glue_catalog/main.tf

@@ -12,14 +12,13 @@ resource "aws_glue_catalog_table" "cleaned_transactions" {
       metadata_operation = "CREATE"
     }
   }
-
-  partition_keys {
+  //Commented because https://github.com/hashicorp/terraform-provider-aws/issues/36531
+  /*partition_keys {
     name = "network_name"
     type = "string"
-  }
+  }*/
 
   parameters = {
-    "table_type"                      = "ICEBERG",
     "write.format.default"            = "parquet"
     "write.parquet.compression-codec" = "zstd"
   }
@@ -133,13 +132,13 @@ resource "aws_glue_catalog_table" "wallets_aggregations" {
     }
   }
 
-  partition_keys {
+  //Commented because https://github.com/hashicorp/terraform-provider-aws/issues/36531
+  /*partition_keys {
     name = "network_name"
     type = "string"
-  }
+  }*/
 
   parameters = {
-    "table_type"                      = "ICEBERG",
     "write.format.default"            = "parquet",
     "write.parquet.compression-codec" = "zstd",
     "write.bucketed-columns"          = "address",
@@ -407,13 +406,13 @@ resource "aws_glue_catalog_table" "features" {
     }
   }
 
-  partition_keys {
+  //Commented because https://github.com/hashicorp/terraform-provider-aws/issues/36531
+  /*partition_keys {
     name = "network_name"
     type = "boolean"
-  }
+  }*/
 
   parameters = {
-    "table_type"                      = "ICEBERG",
     "write.format.default"            = "parquet",
     "write.parquet.compression-codec" = "zstd"
   }

scripts/cloud/etl/orphan_files_removal.py

@@ -20,93 +20,9 @@ def setup_blockchain_db(spark):
 
 
 def setup_iceberg_table(spark):
-    spark.sql("""
-    CREATE TABLE IF NOT EXISTS glue_catalog.bdp.features (
-        block_timestamp FLOAT,
-        block_number FLOAT,
-        transaction_index FLOAT,
-        fee FLOAT,
-        total_transferred_value FLOAT,
-        total_input_value FLOAT,
-        sent_value FLOAT,
-        received_value FLOAT,
-        network_name BOOLEAN,
-
-        avg_sent_value FLOAT,
-        avg_received_value FLOAT,
-        avg_total_value_for_sender FLOAT,
-        avg_total_value_for_receiver FLOAT,
-
-        sum_sent_value FLOAT,
-        sum_received_value FLOAT,
-        sum_total_value_for_sender FLOAT,
-        sum_total_value_for_receiver FLOAT,
-
-        min_sent_value FLOAT,
-        min_received_value FLOAT,
-        min_total_value_for_sender FLOAT,
-        min_total_value_for_receiver FLOAT,
-
-        max_sent_value FLOAT,
-        max_received_value FLOAT,
-        max_total_value_for_sender FLOAT,
-        max_total_value_for_receiver FLOAT,
-
-        median_sent_value FLOAT,
-        median_received_value FLOAT,
-        median_total_value_for_sender FLOAT,
-        median_total_value_for_receiver FLOAT,
-
-        mode_sent_value FLOAT,
-        mode_received_value FLOAT,
-        mode_total_value_for_sender FLOAT,
-        mode_total_value_for_receiver FLOAT,
-
-        stddev_sent_value FLOAT,
-        stddev_received_value FLOAT,
-        stddev_total_value_for_sender FLOAT,
-        stddev_total_value_for_receiver FLOAT,
-
-        num_sent_transactions FLOAT,
-        num_received_transactions FLOAT,
-
-        avg_time_between_sent_transactions FLOAT,
-        avg_time_between_received_transactions FLOAT,
-
-        avg_outgoing_speed_count FLOAT,
-        avg_incoming_speed_count FLOAT,
-        avg_outgoing_speed_value FLOAT,
-        avg_incoming_speed_value FLOAT,
-
-        avg_outgoing_acceleration_count FLOAT,
-        avg_incoming_acceleration_count FLOAT,
-        avg_outgoing_acceleration_value FLOAT,
-        avg_incoming_acceleration_value FLOAT,
-
-        avg_fee_paid FLOAT,
-        total_fee_paid FLOAT,
-        min_fee_paid FLOAT,
-        max_fee_paid FLOAT,
-
-        activity_duration_for_sender FLOAT,
-        first_transaction_timestamp_for_sender FLOAT,
-        last_transaction_timestamp_for_sender FLOAT,
-
-        activity_duration_for_receiver FLOAT,
-        first_transaction_timestamp_for_receiver FLOAT,
-        last_transaction_timestamp_for_receiver FLOAT,
-
-        unique_out_degree FLOAT,
-        unique_in_degree FLOAT
-    )                                                                      
-    PARTITIONED BY (network_name)
-    LOCATION 's3://bdp-features'
-    TBLPROPERTIES (
-        'table_type' = 'ICEBERG',
-        'write.format.default' = 'parquet',
-        'write.parquet.compression-codec' = 'zstd'
-    )
-""")
+    spark.sql(""" ALTER TABLE glue_catalog.bdp.features ADD IF NOT EXISTS
+        PARTITION (network_name = true)         
+        PARTITION (network_name = false) """)
 
 
 cols_dict = {

scripts/cloud/etl/preprocessing.py

@@ -179,26 +179,9 @@ def setup_blockchain_db(spark):
 
 
 def setup_iceberg_table(spark):
-    spark.sql("""
-    CREATE TABLE IF NOT EXISTS glue_catalog.bdp.cleaned_transactions (
-        transaction_id STRING,
-        block_timestamp TIMESTAMP,
-        block_number BIGINT,
-        transaction_hash STRING,
-        transaction_index BIGINT,
-        fee DOUBLE,
-        sender_address STRING,
-        receiver_address STRING,
-        total_transferred_value DOUBLE,
-        total_input_value DOUBLE,
-        sent_value DOUBLE,
-        received_value DOUBLE,
-        network_name STRING
-    )
-    PARTITIONED BY (network_name)
-    LOCATION 's3://bdp-cleaned-transactions'
-    TBLPROPERTIES ('table_type' = 'ICEBERG', 'write.format.default'='parquet', 'write.parquet.compression-codec'='zstd')
-    """)
+    spark.sql("""ALTER TABLE glue_catalog.bdp.cleaned_transactions ADD IF NOT EXISTS
+        PARTITION (network_name = 'ethereum')         
+        PARTITION (network_name = 'bitcoin')""")
 
 def validate_date(date_str: str) -> bool:
     try:

scripts/cloud/etl/transactions_cleaning.py

@@ -161,84 +161,9 @@ def setup_blockchain_db(spark):
     """)
 
 def setup_iceberg_table(spark):
-    spark.sql("""
-    CREATE TABLE IF NOT EXISTS glue_catalog.bdp.wallets_aggregations (
-    address STRING,
-    network_name STRING,
-    
-    avg_sent_value DOUBLE,
-    avg_received_value DOUBLE,
-    avg_total_value_for_sender DOUBLE,
-    avg_total_value_for_receiver DOUBLE,
-
-    sum_sent_value DOUBLE,
-    sum_received_value DOUBLE,
-    sum_total_value_for_sender DOUBLE,
-    sum_total_value_for_receiver DOUBLE,
-
-    min_sent_value DOUBLE,
-    min_received_value DOUBLE,
-    min_total_value_for_sender DOUBLE,
-    min_total_value_for_receiver DOUBLE,
-
-    max_sent_value DOUBLE,
-    max_received_value DOUBLE,
-    max_total_value_for_sender DOUBLE,
-    max_total_value_for_receiver DOUBLE,
-
-    median_sent_value DOUBLE,
-    median_received_value DOUBLE,
-    median_total_value_for_sender DOUBLE,
-    median_total_value_for_receiver DOUBLE,
-
-    mode_sent_value DOUBLE,
-    mode_received_value DOUBLE,
-    mode_total_value_for_sender DOUBLE,
-    mode_total_value_for_receiver DOUBLE,
-
-    stddev_sent_value DOUBLE,
-    stddev_received_value DOUBLE,
-    stddev_total_value_for_sender DOUBLE,
-    stddev_total_value_for_receiver DOUBLE,
-
-    num_sent_transactions BIGINT,
-    num_received_transactions BIGINT,
-
-    avg_time_between_sent_transactions DOUBLE,
-    avg_time_between_received_transactions DOUBLE,
-
-    avg_outgoing_speed_count DOUBLE,
-    avg_incoming_speed_count DOUBLE,
-    avg_outgoing_speed_value DOUBLE,
-    avg_incoming_speed_value DOUBLE,
-
-    avg_outgoing_acceleration_count DOUBLE,
-    avg_incoming_acceleration_count DOUBLE,
-    avg_outgoing_acceleration_value DOUBLE,
-    avg_incoming_acceleration_value DOUBLE,
-
-    avg_fee_paid DOUBLE,
-    total_fee_paid DOUBLE,
-    min_fee_paid DOUBLE,
-    max_fee_paid DOUBLE,
-
-    activity_duration BIGINT,
-    first_transaction_timestamp TIMESTAMP,
-    last_transaction_timestamp TIMESTAMP,
-
-    unique_out_degree BIGINT,
-    unique_in_degree BIGINT
-)
-PARTITIONED BY (network_name)
-LOCATION 's3://bdp-wallets-aggregations'
-TBLPROPERTIES (
-    'table_type' = 'ICEBERG',
-    'write.format.default' = 'parquet',
-    'write.parquet.compression-codec' = 'zstd',
-    'write.bucketed-columns' = 'address',
-    'write.num-buckets' = '2048'
-    )
-    """)
+    spark.sql("""ALTER TABLE glue_catalog.bdp.wallets_aggregations ADD IF NOT EXISTS
+        PARTITION (network_name = 'ethereum')         
+        PARTITION (network_name = 'bitcoin') """)
 
 spark = (
     SparkSession.builder.appName("DataAggregations")