olha00 final pj

Author: yevheniihorbatyuk

encrypted_file/olha00_encrypted_parts.tar.gz

@@ -0,0 +1,35 @@
+import re
+from pathlib import Path
+from pyspark.sql import SparkSession
+from pyspark.sql.functions import udf
+from pyspark.sql.types import StringType
+
+BASE_DIR = Path(__file__).resolve().parent
+BRONZE_DIR = BASE_DIR / "bronze"
+SILVER_DIR = BASE_DIR / "silver"
+
+spark = SparkSession.builder.appName("BronzeToSilverLayer").getOrCreate()
+
+def clean_text(text):
+    return re.sub(r"[^a-zA-Z0-9,.\\\"\' ]", '', str(text))
+
+clean_text_udf = udf(clean_text, StringType())
+
+SILVER_DIR.mkdir(parents=True, exist_ok=True)
+
+df_bio = spark.read.parquet(str(BRONZE_DIR / "athlete_bio"))
+df_results = spark.read.parquet(str(BRONZE_DIR / "athlete_event_results"))
+
+df_bio_cleaned = df_bio.withColumn("name", clean_text_udf(df_bio["name"]))
+df_results_cleaned = df_results.withColumn("event", clean_text_udf(df_results["event"]))
+
+df_bio_cleaned.write.mode("overwrite").parquet(str(SILVER_DIR / "athlete_bio"))
+df_results_cleaned.write.mode("overwrite").parquet(str(SILVER_DIR / "athlete_event_results"))
+
+df_bio_cleaned.show(3)
+df_results_cleaned.show(3)
+
+print(f"Bio rows: {df_bio_cleaned.count()}")
+print(f"Results rows: {df_results_cleaned.count()}")
+
+spark.stop()

olha00/bronze_to_silver.py

@@ -0,0 +1,48 @@
+import requests
+from pyspark.sql import SparkSession
+from pathlib import Path
+
+BASE_DIR = Path(__file__).resolve().parent
+BRONZE_DIR = BASE_DIR / "bronze"
+
+spark = SparkSession.builder.appName("LandingToBronzeLayer").getOrCreate()
+
+
+def download_data(local_file_path):
+    url = "https://ftp.goit.study/neoversity/"
+    downloading_url = url + local_file_path + ".csv"
+    print(f"Downloading: {downloading_url}")
+    response = requests.get(downloading_url)
+
+    if response.status_code == 200:
+        save_path = BRONZE_DIR / f"{local_file_path}.csv"
+        with open(save_path, "wb") as file:
+            file.write(response.content)
+        print(f"Saved: {save_path}")
+    else:
+        print(f"Failed: {local_file_path} (Code: {response.status_code})")
+
+
+def main():
+    BRONZE_DIR.mkdir(parents=True, exist_ok=True)  # Create folder
+
+    files = ["athlete_bio", "athlete_event_results"]
+
+    for filename in files:
+        download_data(filename)
+
+
+    for filename in files:
+        csv_path = BRONZE_DIR / f"{filename}.csv"
+        df = spark.read.option("header", True).csv(str(csv_path))
+        print(f"Preview {filename}:")
+        df.show(3)
+        print(f"Rows: {df.count()}")
+
+        df.write.mode("overwrite").parquet(str(BRONZE_DIR / filename))
+        print(f"Parquet saved: {BRONZE_DIR / filename}")
+
+if __name__ == "__main__":
+    main()
+
+spark.stop()

olha00/landing_to_bronze.py

@@ -0,0 +1,45 @@
+import os
+from datetime import datetime
+from airflow import DAG
+from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator
+
+
+BASE_DIR = os.path.dirname(os.path.abspath(__file__))
+
+default_args = {
+    "owner": "airflow",
+    "start_date": datetime(2025, 12, 10),
+    "depends_on_past": False,
+    "retries": 1,
+}
+
+with DAG(
+    dag_id="ola_de_fp2",
+    default_args=default_args,
+    schedule_interval=None,
+    catchup=False,
+    description="ETL pipeline from landing to gold using Spark and Airflow",
+) as dag:
+
+    landing_to_bronze = SparkSubmitOperator(
+        task_id="ola_landing_to_bronze",
+        application=os.path.join(BASE_DIR, "landing_to_bronze.py"),
+        conn_id="spark-default",
+        verbose=True,
+    )
+
+    bronze_to_silver = SparkSubmitOperator(
+        task_id="ola_bronze_to_silver",
+        application=os.path.join(BASE_DIR, "bronze_to_silver.py"),
+        conn_id="spark-default",
+        verbose=True,
+    )
+
+    silver_to_gold = SparkSubmitOperator(
+        task_id="ola_silver_to_gold",
+        application=os.path.join(BASE_DIR, "silver_to_gold.py"),
+        conn_id="spark-default",
+        verbose=True,
+    )
+
+    landing_to_bronze >> bronze_to_silver >> silver_to_gold

olha00/project_solution.py

@@ -0,0 +1,34 @@
+from pyspark.sql import SparkSession
+from pyspark.sql.functions import avg, current_timestamp
+from pathlib import Path
+
+BASE_DIR = Path(__file__).resolve().parent
+SILVER_DIR = BASE_DIR / "silver"
+GOLD_DIR = BASE_DIR / "gold"
+
+spark = SparkSession.builder.appName("SilverToGoldLayer").getOrCreate()
+
+GOLD_DIR.mkdir(parents=True, exist_ok=True)
+
+df_bio = spark.read.parquet(str(SILVER_DIR / "athlete_bio"))
+df_results = spark.read.parquet(str(SILVER_DIR / "athlete_event_results"))
+
+df_joined = df_results.join(df_bio, on="athlete_id", how="inner")
+
+df_avg = df_joined.groupBy(
+    "sport",
+    "medal",
+    "sex",
+    df_bio["country_noc"]
+).agg(
+    avg("weight").alias("avg_weight"),
+    avg("height").alias("avg_height")
+).withColumn(
+    "timestamp", current_timestamp()
+)
+
+df_avg.show()
+
+df_avg.write.mode("overwrite").parquet(str(GOLD_DIR / "avg_stats"))
+
+spark.stop()