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()
Note : you need to fix the Indentation of the code i shared as it is sensitive to indentation
When i paste it here it looses the Indentation
Or you can download the testjob.py from here
https://github.com/puneetvijwani/fabricNotebooks
Perfect, that works. Thanks again for your extensive speeedy help on this, much appreciated. 🙏
Have a super day.