intermediate
File Streaming
8 min readLast updated: 2026-07-09
Overview
Learn how to read files written to directories in real-time as a continuous data stream, and configure Spark's folder listeners.
What You Will Learn
In this lesson, you will learn:
- Directory Source: Reading new files written to folders.
- File formats: Streaming CSV, JSON, and Parquet directories.
- Clean schemas: Defining schemas for directory files.
Detailed Concept Explanation
You can configure Spark to monitor a folder path and read new files as they are added. This is a very common way to process data streams when upstream systems upload batch files to cloud storage buckets (like Amazon S3 or Google Cloud Storage).
Key settings:
schema: Explicitly defines file structure columns. Spark requires schema declarations for file streaming sources.maxFilesPerTrigger: Specifies the maximum number of new files to process in a single micro-batch trigger, preventing memory overloads.
Code Examples
Python (PySpark) Implementation
python
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, IntegerType
spark = SparkSession.builder.appName("FileStreaming").getOrCreate()
# Define schema for input files
user_schema = StructType([
StructField("id", IntegerType(), True),
StructField("name", StringType(), True),
StructField("city", StringType(), True)
])
# Read streaming JSON files from directory
file_stream = spark.readStream \
.schema(user_schema) \
.json("input_dir")
# Write stream to console
query = file_stream.writeStream.format("console").start()
query.awaitTermination(5)
query.stop()
Expected Output
text
+---+-----+------+
| id| name| city|
+---+-----+------+
| 1|Alice|London|
+---+-----+------+
Execution Plan Diagram (Python & Scala)
Execution Plan Diagram
SparkSession.builder
StructType
readStream.schema().json()
writeStream.format(console)
start()
awaitTermination(5)
stop()
Scala Implementation
scala
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types._
val spark = SparkSession.builder().appName("FileStreamingScala").getOrCreate()
val userSchema = StructType(Array(
StructField("id", IntegerType, true),
StructField("name", StringType, true),
StructField("city", StringType, true)
))
val fileStream = spark.readStream
.schema(userSchema)
.json("input_dir")
val query = fileStream.writeStream.format("console").start()
query.awaitTermination(5000)
query.stop()
Expected Output
text
+---+-----+------+
| id| name| city|
+---+-----+------+
| 1|Alice|London|
+---+-----+------+
Common Mistakes
- Assuming schema inference works natively: Leaving out the schema definition when reading directory files. Structured Streaming requires explicit schemas for file sources to maintain metadata consistency.
Best Practices
- Use maxFilesPerTrigger: Always configure the
maxFilesPerTriggeroption to control backpressure during folder reads.