Window functions और streaming queries

Databricks में Spark SQL के साथ Data Transformation

Disha Mukherjee

Lead Data Engineer

जब groupBy काफ़ी नहीं होता

recraft: half: स्प्रेडशीट डेटा की पंक्तियों पर आवर्धक काँच, जिनमें गणना किए गए मान हाइलाइट हैं

groupBy() → प्रति समूह 1 पंक्ति (totals, averages)
Window function → एक कॉलम जोड़ता है, सभी पंक्तियाँ रखता है
Use case: running totals, rankings, row comparisons

Running total

from pyspark.sql.window import Window

window_spec = (
    Window.partitionBy("Customer_ID")
    .orderBy("Date")
    .rowsBetween(Window.unboundedPreceding, Window.currentRow)
)


df_running = df_valid.withColumn(
    "running_total",
    F.round(F.sum("Transaction_Amount").over(window_spec), 2)
)

Running total

+-----------+-------------------+------------------+-------------+
|Customer_ID|Date               |Transaction_Amount|running_total|
+-----------+-------------------+------------------+-------------+
|CUST003    |2023-01-03 00:00:00|5752.36           |5752.36      |
|CUST003    |2023-02-14 00:00:00|12340.00          |18092.36     |
+-----------+-------------------+------------------+-------------+

ग्राहकों की रैंकिंग

customer_totals = (
    df_valid.groupBy("Customer_ID")
    .agg(F.round(F.sum("Transaction_Amount"), 2).alias("total_revenue"))
)


rank_window = Window.orderBy(F.col("total_revenue").desc())

df_ranked = customer_totals.withColumn("revenue_rank", F.rank().over(rank_window))

+-----------+-------------+------------+
|Customer_ID|total_revenue|revenue_rank|
+-----------+-------------+------------+
|CUST1469   |99996.20     |1           |
|CUST19129  |99990.14     |2           |
|CUST39417  |99982.21     |3           |
+-----------+-------------+------------+

Streaming क्या है?

Batch - पूरे fixed डेटासेट को एक साथ प्रोसेस करें
Streaming - डेटा आते ही incrementally प्रोसेस करें
ट्रांज़ैक्शन फ़ीड, logs, और event डेटा के लिए उपयुक्त

recraft: half: तीरों के साथ एक प्रोसेसिंग इंजन में लगातार बहता डेटा स्ट्रीम के रूप में

फ़ाइल-आधारित streaming

Stream directory:
  day_1.csv  (108 KB)
  day_2.csv  (108 KB)
  day_3.csv  (108 KB)
  day_4.csv  (108 KB)
  day_5.csv  (108 KB)

CSV फ़ाइलों की डायरेक्टरी = streaming source
हर नई फ़ाइल = एक micro-batch
नई फ़ाइलें अपने-आप pick हो जाती हैं
Schema स्पष्ट रूप से define करना होगा

स्ट्रीम पढ़ना

df_stream = (
    spark.readStream.format("csv")
    .option("header", "true")
    .schema(streaming_schema)
    .load(STREAM_DIR)
)

print(df_stream.isStreaming)

True

Checkpoints क्या हैं?

recraft: half: एक बहती डेटा पाइपलाइन में बुकमार्क/सेव-पॉइंट मार्कर, प्रोग्रेस सेव है

Checkpoint = डिस्क पर metadata डायरेक्टरी
कौन-सी फ़ाइलें प्रोसेस हुईं, इसका record रखता है
रीस्टार्ट पर Spark वहीं से फिर शुरू करता है
डुप्लिकेट प्रोसेसिंग रोकता है

स्ट्रीम लिखना

query = (
    df_stream.writeStream.format("delta")
    .outputMode("append")
    .option("checkpointLocation", CHECKPOINT_DIR)
    .option("path", DELTA_PATH)
    .trigger(availableNow=True)
    .start()
)
query.awaitTermination()

Status:       Stopped
Rows written: 5,000

मॉनिटरिंग: status और lastProgress

print(query.status)

progress = query.lastProgress
print(f"Rows processed: {progress['numInputRows']}")
print(f"Rows/sec: {progress['processedRowsPerSecond']:.0f}")

{'message': 'Stopped', 'isDataAvailable': False, 'isTriggerActive': False}
Rows processed: 5,000
Rows/sec:       752

Checkpoint रिकवरी

query_restart = (
    df_stream.writeStream.format("delta")
    .option("checkpointLocation", CHECKPOINT_DIR)
    .option("path", DELTA_PATH)
    .trigger(availableNow=True)
    .start()
)
query_restart.awaitTermination()

Rows on restart: 0

अभ्यास करते हैं!

Databricks में Spark SQL के साथ Data Transformation