# Databricks notebook source
# MAGIC %md
# MAGIC # Nordvella: pierwsze kroki w Databricks
# MAGIC Notebook do artykułu „Czym jest Databricks - kompletny przewodnik” (blog JSystems).
# MAGIC Zanim uruchomisz komórki, wgraj pliki CSV z artykułu do wolumenu `workspace.nordvella_lakehouse.raw_files`.

# COMMAND ----------

# MAGIC %sql
# MAGIC CREATE SCHEMA IF NOT EXISTS workspace.nordvella_lakehouse;
# MAGIC CREATE VOLUME IF NOT EXISTS workspace.nordvella_lakehouse.raw_files;

# COMMAND ----------

path = "/Volumes/workspace/nordvella_lakehouse/raw_files/"

sales = (spark.read
         .option("header", True)
         .option("inferSchema", True)
         .csv(path + "nordvella_sales.csv"))

print(f"Liczba wierszy: {sales.count()}")
sales.printSchema()

# COMMAND ----------

display(sales.limit(10))

# COMMAND ----------

# Zapis trzech plików jako tabele Delta w Unity Catalog
for name in ["sales", "stores", "products"]:
    df = (spark.read
          .option("header", True)
          .option("inferSchema", True)
          .csv(f"{path}nordvella_{name}.csv"))
    df.write.mode("overwrite").saveAsTable(f"workspace.nordvella_lakehouse.{name}")
    print(name, df.count())

# COMMAND ----------

# MAGIC %sql
# MAGIC -- Sprzedaż netto w tysiącach zł, miesięcznie, w podziale na regiony
# MAGIC SELECT CAST(date_trunc('MONTH', s.sale_date) AS DATE) AS month,
# MAGIC        st.region,
# MAGIC        ROUND(SUM(s.net_amount) / 1000, 1) AS net_sales_k_pln
# MAGIC FROM workspace.nordvella_lakehouse.sales AS s
# MAGIC JOIN workspace.nordvella_lakehouse.stores AS st USING (store_id)
# MAGIC WHERE s.is_returned = 0
# MAGIC GROUP BY ALL
# MAGIC ORDER BY month, st.region

# COMMAND ----------

from pyspark.sql import functions as F

sales = spark.table("workspace.nordvella_lakehouse.sales")
products = spark.table("workspace.nordvella_lakehouse.products").select("product_id", "category")

by_category = (sales.where("is_returned = 0")
               .join(products, "product_id")
               .groupBy("category")
               .agg(F.round(F.sum("net_amount") / 1e6, 2).alias("net_sales_m_pln"),
                    F.round(F.sum("net_margin") / F.sum("net_amount") * 100, 1).alias("margin_pct"))
               .orderBy(F.desc("net_sales_m_pln")))

display(by_category)