import sysfrom awsglue.context import GlueContextglueContext = GlueContext(sc)dyf = glueContext.create_dynamic_frame.from_catalog(database="lake_raw", table_name="clickstream")df = dyf.toDF().select("session_id", "page_url")glueContext.write_dynamic_frame.from_options(frame=dyf, connection_type="s3", connection_options={"path": "s3://acme-lake/warehouse/web.db/sessions/"}, format="parquet")%pip install xdf = spark.read.json("s3://lake/events/app_clicks/2024/06/01/")df.createOrReplaceTempView("clicks")%%sqlSELECT user_id, count(*) FROM clicks GROUP BY user_idimport boto3, pandas as pdimport snowflake.connectors3 = boto3.client("s3")obj = s3.get_object(Bucket="acme-landing", Key="vendors/acme_supply/invoices/2024/05/invoices_20240501.csv")df = pd.read_csv("s3://acme-landing/vendors/acme_supply/shipments/shipments.csv", usecols=["ship_id", "ship_date"])conn = snowflake.connector.connect(user="x", account="y")cur = conn.cursor()schema = "SALES"QUERY = f"""SELECT i.invoice_id, i.amount, v.vendor_nameFROM RAW_DB.{schema}.INVOICES iLEFT JOIN RAW_DB.{schema}.VENDORS v ON i.vendor_id = v.vendor_idWHERE i.invoice_date >= %s"""cur.execute(QUERY, ("2024-01-01",))pd.read_sql("select order_id, sku from RAW_DB.SALES.ORDER_LINES where qty > 0", conn)df.to_parquet("s3://acme-curated/vendors/shipments_clean/")USE WAREHOUSE etl_wh;-- daily customer revenueCREATE OR REPLACE TABLE ANALYTICS.MART.CUSTOMER_REVENUE ASWITH orders AS ( SELECT o.order_id, o.customer_id AS cust_id, o.order_total, o.order_ts::date AS order_date FROM RAW_DB.SALES.ORDERS o WHERE o.status <> 'CANCELLED'),cust AS ( SELECT c.customer_id, c.first_name || ' ' || c.last_name AS full_name, c.region FROM RAW_DB.CRM.CUSTOMERS c)SELECT cu.full_name, cu.region, SUM(order_total) AS revenue, COUNT(*) AS n_ordersFROM ordersJOIN cust cu ON orders.cust_id = cu.customer_idGROUP BY 1, 2QUALIFY ROW_NUMBER() OVER (PARTITION BY region ORDER BY revenue DESC) <= 10;MERGE INTO ANALYTICS.MART.DIM_PRODUCT tUSING (SELECT product_id, product_name, category FROM RAW_DB.CATALOG.PRODUCTS) sON t.product_id = s.product_idWHEN MATCHED THEN UPDATE SET t.product_name = s.product_name, t.category = s.categoryWHEN NOT MATCHED THEN INSERT (product_id, product_name, category) VALUES (s.product_id, s.product_name, s.category);COPY INTO RAW_DB.SALES.ORDERS_STGFROM @RAW_DB.SALES.S3_STAGE/orders/year=2024/FILE_FORMAT = (TYPE = PARQUET);INSERT INTO ANALYTICS.MART.EVENTS (event_id, user_id, payload_type)SELECT e.id, e.user_id, e.payload:type::stringFROM RAW_DB.APP.EVENTS e, LATERAL FLATTEN(input => e.payload:items) fWHERE f.value:sku IS NOT NULL;UPDATE ANALYTICS.MART.DIM_PRODUCT SET is_active = FALSE WHERE last_seen < DATEADD(day, -90, CURRENT_DATE);SELECT * FROM {{ ref('stg_payments') }} p JOIN {{ source('stripe','charges') }} ch ON p.charge_id = ch.id;CREATE OR REPLACE PROCEDURE ANALYTICS.MART.REFRESH_X()RETURNS STRING LANGUAGE SQL AS$$BEGIN INSERT INTO ANALYTICS.MART.X SELECT a, b FROM RAW_DB.SALES.Y; RETURN 'ok';END;$$;from pyspark.sql import SparkSession, functions as Ffrom pyspark.sql.functions import colspark = SparkSession.builder.appName("orders").getOrCreate()BUCKET = "acme-datalake-prod"env = "prod"orders_path = f"s3a://{BUCKET}/curated/sales/orders/"orders = spark.read.parquet(orders_path)customers = (spark.read.format("csv") .option("header", "true") .load("s3://acme-datalake-prod/raw/crm/customers/dt=2024-05-01/customers.csv"))returns = spark.table("finance_db.returns")joined = (orders.filter(col("status") == "SHIPPED") .join(customers, on="customer_id", how="left") .select("order_id", "customer_id", F.col("order_total"), customers["region"]) .withColumn("net_total", F.col("order_total") - F.coalesce(F.col("discount"), F.lit(0))))joined.createOrReplaceTempView("v_orders")spark.sql(""" SELECT region, SUM(net_total) AS revenue FROM v_orders GROUP BY region""").createOrReplaceTempView("v_rev")rev = spark.sql(f"SELECT * FROM {env}_analytics.region_rev r JOIN v_rev ON r.region = v_rev.region")returns.groupBy("return_reason").agg(F.sum("refund_amt").alias("refunds"))(joined.write.mode("overwrite") .partitionBy("region") .parquet(f"s3://{BUCKET}/gold/sales/order_net/"))rev.write.saveAsTable("analytics.region_revenue")sf_opts = {"sfDatabase": "RAW_DB", "sfSchema": "SALES"}inv = (spark.read.format("snowflake").options(**sf_opts) .option("dbtable", "RAW_DB.SALES.INVENTORY").load())inv.selectExpr("sku", "qty_on_hand * unit_cost AS stock_value")keep_cols = ["sku", "warehouse_id", "qty_on_hand"]inv.select(*keep_cols).write.parquet(f"s3://{BUCKET}/gold/inventory/inventory_snapshot/")