Forum Discussion
Spark Job Structured Streaming Lakehouse Files
- 2 years ago
BilalBobat lets try this.. more simpler version, This works for me just Fine
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType
spark = SparkSession.builder \
.appName("Stream CSV to Delta Table") \
.getOrCreate()
userSchema = StructType().add("name", "string").add("sales", "integer")
streamingDF = spark.readStream \
.schema(userSchema) \
.option("maxFilesPerTrigger", 1) \
.csv("Files/Streaming/") # Replace with the actual path to your streaming CSV files
query = streamingDF.writeStream \
.trigger(processingTime='5 seconds') \
.outputMode("append") \
.format("delta") \
.option("checkpointLocation", "Tables/Streaming_Table_test/_checkpoint") \
.start("Tables/Streaming_Table_test") # Replace with the path where you want to save the Delta table
query.awaitTermination()
BilalBobat lets try this.. more simpler version, This works for me just Fine
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType
spark = SparkSession.builder \
.appName("Stream CSV to Delta Table") \
.getOrCreate()
userSchema = StructType().add("name", "string").add("sales", "integer")
streamingDF = spark.readStream \
.schema(userSchema) \
.option("maxFilesPerTrigger", 1) \
.csv("Files/Streaming/") # Replace with the actual path to your streaming CSV files
query = streamingDF.writeStream \
.trigger(processingTime='5 seconds') \
.outputMode("append") \
.format("delta") \
.option("checkpointLocation", "Tables/Streaming_Table_test/_checkpoint") \
.start("Tables/Streaming_Table_test") # Replace with the path where you want to save the Delta table
query.awaitTermination()