GPT

Kafka Streaming Pipeline

from pyspark.sql.functions import fromjson, col, currenttimestamp from pyspark.sql.types import StructType, StringType, DoubleType, TimestampType orderschema = StructType() \ .add("orderid", StringType()) \ .add("cust

Prompt text

Original English text. Replace placeholders with your own details.

from pyspark.sql.functions import from_json, col, current_timestamp
from pyspark.sql.types import StructType, StringType, DoubleType, TimestampType

order_schema = StructType() \
 .add("order_id", StringType()) \
 .add("customer_id", StringType()) \
 .add("revenue", DoubleType()) \
 .add("event_time", TimestampType())

def stream_bronze_orders(kafka_bootstrap: str, topic: str, bronze_path: str):
 stream = spark.readStream \
 .format("kafka") \
 .option("kafka.bootstrap.servers", kafka_bootstrap) \
 .option("subscribe", topic) \
 .option("startingOffsets", "latest") \
 .option("failOnDataLoss", "false") \
 .load()

 parsed = stream.select(
 from_json(col("value").cast("string"), order_schema).alias("data"),
 col("timestamp").alias("_kafka_timestamp"),
 current_timestamp().alias("_ingested_at")
 ).select("data.*", "_kafka_timestamp", "_ingested_at")

 return parsed.writeStream \
 .format("delta") \
 .outputMode("append") \
 .option("checkpointLocation", f"{bronze_path}/_checkpoint") \
 .option("mergeSchema", "true") \
 .trigger(processingTime="30 seconds") \
 .start(bronze_path)

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 →