GPT

Spark Pipeline (pyspark + Delta Lake)

from pyspark.sql import SparkSession from pyspark.sql.functions import col, currenttimestamp, sha2, concatws, lit from delta.tables import DeltaTable spark = SparkSession.builder \ .config("spark.sql.extensions", "io.d

Prompt text

Original English text. Replace placeholders with your own details.

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, current_timestamp, sha2, concat_ws, lit
from delta.tables import DeltaTable

spark = SparkSession.builder \
 .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
 .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
 .getOrCreate()

# ── Bronze: raw ingest (append-only, schema-on-read) ─────────────────────────
def ingest_bronze(source_path: str, bronze_table: str, source_system: str) -> int:
 df = spark.read.format("json").option("inferSchema", "true").load(source_path)
 df = df.withColumn("_ingested_at", current_timestamp()) \
 .withColumn("_source_system", lit(source_system)) \
 .withColumn("_source_file", col("_metadata.file_path"))
 df.write.format("delta").mode("append").option("mergeSchema", "true").save(bronze_table)
 return df.count()

# ── Silver: cleanse, deduplicate, conform ────────────────────────────────────
def upsert_silver(bronze_table: str, silver_table: str, pk_cols: list[str]) -> None:
 source = spark.read.format("delta").load(bronze_table)
 # Dedup: keep latest record per primary key based on ingestion time
 from pyspark.sql.window import Window
 from pyspark.sql.functions import row_number, desc
 w = Window.partitionBy(*pk_cols).orderBy(desc("_ingested_at"))
 source = source.withColumn("_rank", row_number().over(w)).filter(col("_rank") == 1).drop("_rank")

 if DeltaTable.isDeltaTable(spark, silver_table):
 target = DeltaTable.forPath(spark, silver_table)
 merge_condition = " AND ".join([f"target.{c} = source.{c}" for c in pk_cols])
 target.alias("target").merge(source.alias("source"), merge_condition) \
 .whenMatchedUpdateAll() \
 .whenNotMatchedInsertAll() \
 .execute()
 else:
 source.write.format("delta").mode("overwrite").save(silver_table)

# ── Gold: aggregated business metric ─────────────────────────────────────────
def build_gold_daily_revenue(silver_orders: str, gold_table: str) -> None:
 df = spark.read.format("delta").load(silver_orders)
 gold = df.filter(col("status") == "completed") \
 .groupBy("order_date", "region", "product_category") \
 .agg({"revenue": "sum", "order_id": "count"}) \
 .withColumnRenamed("sum(revenue)", "total_revenue") \
 .withColumnRenamed("count(order_id)", "order_count") \
 .withColumn("_refreshed_at", current_timestamp())
 gold.write.format("delta").mode("overwrite") \
 .option("replaceWhere", f"order_date >= '{gold['order_date'].min()}'") \
 .save(gold_table)

Source & attribution

Awesome Prompts · Awesome Prompts

Source-listed license: GPL-3.0. Prompt text is preserved; titles and previews may be shortened for display. Source examples are not NexGateHub test results.

View original source ↗ · Source credits & licenses

Explore more prompts →