Features/bdp 94 lof (#37)

* refactor in local folder

* lof poc

Author: WikarNotAvailable

scripts/local/aggregate/aggregate.py

@@ -0,0 +1,18 @@
+from pyspark.sql import SparkSession
+from scripts.local.shared.schemas import transaction_schema
+from scripts.local.aggregate.components.aggregations import aggregate
+
+spark = (
+    SparkSession.builder.appName("DataAggregations")    
+    .config("spark.sql.parquet.enableVectorizedReader", "true")
+    .config("spark.sql.parquet.mergeSchema", "false") # No need as we explicitly specify the schema
+    .config("spark.executor.memory", "6g")
+    .config("spark.driver.memory", "2g")
+    # .config("spark.local.dir", "/mnt/d/spark-temp") # Change the temp directory
+    .getOrCreate()
+)
+
+source_dir = "data/historical/etl/transactions"
+output_dir = "data/historical/aggregations" 
+
+aggregate(spark, source_dir, output_dir, transaction_schema)

scripts/local/aggregate/aggregate_benchmark.py

@@ -0,0 +1,18 @@
+from pyspark.sql import SparkSession
+from scripts.local.shared.benchmark_schemas import benchmark_transaction_schema
+from components.aggregations import aggregate
+
+spark = (
+    SparkSession.builder.appName("DataAggregations")    
+    .config("spark.sql.parquet.enableVectorizedReader", "true")
+    .config("spark.sql.parquet.mergeSchema", "false") # No need as we explicitly specify the schema
+    .config("spark.executor.memory", "6g")
+    .config("spark.driver.memory", "2g")
+    # .config("spark.local.dir", "/mnt/d/spark-temp") # Change the temp directory
+    .getOrCreate()
+)
+
+source_dir = "data/benchmark/etl/transactions" 
+output_dir = "data/benchmark/aggregations" 
+
+aggregate(spark, source_dir, output_dir, benchmark_transaction_schema)

scripts/local/etl/aggregations.py …cal/aggregate/components/aggregations.pyscripts/local/etl/aggregations.py renamed to scripts/local/aggregate/components/aggregations.py scripts/local/etl/aggregations.py renamed to scripts/local/aggregate/components/aggregations.py

@@ -1,7 +1,6 @@
 from pyspark.sql.functions import mean, mode, stddev, count, median, sum, min, max, col, lit, count_distinct, unix_timestamp, lag, first, when, monotonically_increasing_id
-from pyspark.sql import SparkSession, DataFrame
+from pyspark.sql import DataFrame
 from pyspark.sql.window import Window
-from scripts.local.shared.schemas import transaction_schema
 
 def calculate_aggregations(df):
     sender_window = Window.partitionBy("sender_address").orderBy("block_timestamp")
@@ -147,39 +146,26 @@ def preprocess_btc_df(df):
 
     return df_btc_send.unionByName(df_btc_receive)
 
-    
-spark = (
-    SparkSession.builder.appName("DataAggregations")    
-    .config("spark.sql.parquet.enableVectorizedReader", "true")
-    .config("spark.sql.parquet.mergeSchema", "false") # No need as we explicitly specify the schema
-    .config("spark.executor.memory", "6g")
-    .config("spark.driver.memory", "2g")
-    # .config("spark.local.dir", "/mnt/d/spark-temp") # Change the temp directory
-    .getOrCreate()
-)
-
-source_dir = "data/historical/etl/transactions" #data/benchmark/etl/transactions or data/historical/etl/transactions
-output_dir = "data/historical/aggregations" #data/benchmark/aggregations or data/historical/aggregations
-
-cols_to_drop = ["transaction_id", "block_number", "transaction_index"]
-transaction_df = spark.read.schema(transaction_schema).parquet(source_dir).drop(*cols_to_drop)
+def aggregate(spark, source_dir, output_dir, schema):
+    cols_to_drop = ["transaction_id", "block_number", "transaction_index"]
+    transaction_df = spark.read.schema(schema).parquet(source_dir).drop(*cols_to_drop)
 
-unique_degrees_df = calculate_unique_degrees(transaction_df)
+    unique_degrees_df = calculate_unique_degrees(transaction_df)
 
-df_eth = transaction_df.where(col("network_name") == "ethereum")
-df_btc = transaction_df.where(col("network_name") == "bitcoin")
+    df_eth = transaction_df.where(col("network_name") == "ethereum")
+    df_btc = transaction_df.where(col("network_name") == "bitcoin")
 
-df_btc = preprocess_btc_df(df_btc)
+    df_btc = preprocess_btc_df(df_btc)
 
-df_btc_aggregations = calculate_aggregations(df_btc)
-df_eth_aggregations = calculate_aggregations(df_eth)
+    df_btc_aggregations = calculate_aggregations(df_btc)
+    df_eth_aggregations = calculate_aggregations(df_eth)
 
-df_btc_aggregations = df_btc_aggregations.withColumn("network_name", lit("bitcoin"))
-df_eth_aggregations = df_eth_aggregations.withColumn("network_name", lit("ethereum"))
+    df_btc_aggregations = df_btc_aggregations.withColumn("network_name", lit("bitcoin"))
+    df_eth_aggregations = df_eth_aggregations.withColumn("network_name", lit("ethereum"))
 
-aggregations_df = df_btc_aggregations.unionByName(df_eth_aggregations)
-aggregations_df = aggregations_df.join(unique_degrees_df, "address", "outer").na.fill(0)
+    aggregations_df = df_btc_aggregations.unionByName(df_eth_aggregations)
+    aggregations_df = aggregations_df.join(unique_degrees_df, "address", "outer").na.fill(0)
 
-aggregations_df.coalesce(1).write.parquet(output_dir, mode="overwrite", compression="zstd")
+    aggregations_df.coalesce(1).write.parquet(output_dir, mode="overwrite", compression="zstd")
 
-spark.stop()
+    spark.stop()

scripts/local/anomalies_detection/preprocessing/components/preprocess_data_second_step.py

@@ -1,7 +1,7 @@
 from functools import reduce
 from typing import Callable
 from pyspark.sql import SparkSession, DataFrame
-from scripts.local.shared.schemas import benchmark_input_schema
+from scripts.local.shared.benchmark_schemas import benchmark_input_schema
 
 def extract(spark: SparkSession, path: str) -> DataFrame:
     df = spark.read.schema(benchmark_input_schema).csv(path, header=True)

scripts/local/anomalies_detection/preprocessing/preprocess_benchmark_datasets.py

@@ -1,9 +1,8 @@
 from pyspark.sql.functions import col
-from scripts.local.shared.schemas import transaction_scaled_schema, aggregations_scaled_schema
 from scripts.local.shared.consts import sender_fields, receiver_fields, common_fields
 
-def join_transactions_with_aggregations(spark, transactions_dir, aggregations_dir, output_dir):
-    transactions_df = spark.read.schema(transaction_scaled_schema).parquet(transactions_dir)
+def join_transactions_with_aggregations(spark, transactions_dir, aggregations_dir, output_dir, transactions_scaled_schema, aggregations_scaled_schema):
+    transactions_df = spark.read.schema(transactions_scaled_schema).parquet(transactions_dir)
     aggregations_df = spark.read.schema(aggregations_scaled_schema).parquet(aggregations_dir)
 
     sender_aggregations = aggregations_df.select(
@@ -30,4 +29,4 @@ def join_transactions_with_aggregations(spark, transactions_dir, aggregations_di
         compression="zstd"
     )
 
-    return final_df
+    return final_df