Eric Campbell 5 min read Python

PySpark - Create RDD from Text File

Resilient Distributed Datasets (RDDs) represent PySpark's fundamental abstraction for distributed data processing. While DataFrames have become the preferred API for structured data, RDDs remain...

Key Insights

  • Use SparkContext.textFile() to load text files into RDDs, with each line becoming a separate element in the distributed dataset—this method automatically handles file partitioning and supports wildcards, directories, and compressed formats.
  • Control partition count explicitly using the minPartitions parameter to optimize performance; too few partitions underutilize your cluster while too many create excessive overhead.
  • For production workloads processing large text files, always cache RDDs when performing multiple operations and consider migrating to DataFrames for structured data processing to leverage Catalyst optimizer benefits.

Introduction to RDDs and Text File Processing

Resilient Distributed Datasets (RDDs) represent PySpark’s fundamental abstraction for distributed data processing. While DataFrames have become the preferred API for structured data, RDDs remain essential for processing unstructured text files where you need fine-grained control over data transformations.

Text files are ubiquitous in data engineering pipelines. You’ll encounter them in log analysis, ETL preprocessing, natural language processing, and legacy system integrations. Unlike structured formats, text files require custom parsing logic, making RDDs’ low-level transformation capabilities particularly valuable.

The beauty of creating RDDs from text files lies in PySpark’s ability to automatically distribute file reading across your cluster. Whether you’re processing a 10MB log file or a 10TB dataset, the API remains identical while PySpark handles the complexity of distributed computation.

Setting Up SparkContext

Before creating RDDs, you need an active SparkContext. This object serves as your connection to the Spark cluster and provides methods for creating RDDs from various data sources.

from pyspark import SparkContext, SparkConf

# Configure Spark application
conf = SparkConf().setAppName("TextFileRDDExample").setMaster("local[*]")
sc = SparkContext(conf=conf)

# Verify SparkContext is active
print(f"Spark Version: {sc.version}")
print(f"Application ID: {sc.applicationId}")

For modern PySpark applications, you might use SparkSession instead, which provides a unified entry point:

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("TextFileRDDExample") \
    .master("local[*]") \
    .getOrCreate()

# Access SparkContext from SparkSession
sc = spark.sparkContext

The local[*] master URL runs Spark locally using all available CPU cores—perfect for development and testing. In production, you’d specify your cluster manager URL (YARN, Mesos, or Kubernetes).

Basic RDD Creation with textFile()

The textFile() method is your primary tool for loading text data into RDDs. Each line in the file becomes a separate element in the RDD.

# Read a single text file
log_rdd = sc.textFile("data/application.log")

# Display first 5 lines
for line in log_rdd.take(5):
    print(line)

# Count total lines
print(f"Total lines: {log_rdd.count()}")

Reading multiple files from a directory is equally straightforward:

# Read all text files in a directory
all_logs_rdd = sc.textFile("data/logs/*.txt")

# Or specify the directory directly (reads all files)
directory_rdd = sc.textFile("data/logs/")

# Read multiple specific files using comma separation
multi_file_rdd = sc.textFile("data/file1.txt,data/file2.txt")

PySpark supports wildcards and glob patterns, making it easy to filter files by pattern:

# Read only files matching a pattern
filtered_rdd = sc.textFile("data/logs/app-2024-*.log")

Working with RDD Partitions

Partitioning determines how your data is distributed across the cluster. Proper partition management is crucial for performance optimization.

By default, PySpark creates one partition per HDFS block (typically 128MB). You can override this with the minPartitions parameter:

# Create RDD with minimum 8 partitions
partitioned_rdd = sc.textFile("data/large_file.txt", minPartitions=8)

# Check actual partition count
print(f"Number of partitions: {partitioned_rdd.getNumPartitions()}")

The minPartitions parameter is a suggestion, not a guarantee. PySpark may create more partitions based on file size and block boundaries.

Repartitioning allows you to adjust partition count after RDD creation:

# Read file with default partitioning
data_rdd = sc.textFile("data/input.txt")
print(f"Initial partitions: {data_rdd.getNumPartitions()}")

# Increase partitions for better parallelism
repartitioned_rdd = data_rdd.repartition(16)
print(f"After repartition: {repartitioned_rdd.getNumPartitions()}")

# Decrease partitions to reduce overhead
coalesced_rdd = data_rdd.coalesce(2)
print(f"After coalesce: {coalesced_rdd.getNumPartitions()}")

Use repartition() when increasing partition count or when you need data shuffled. Use coalesce() when decreasing partitions—it’s more efficient because it avoids a full shuffle.

Common RDD Transformations on Text Data

Once you’ve loaded text data into an RDD, you’ll typically apply transformations to extract meaningful information.

Filtering lines based on conditions is a fundamental operation:

# Filter log lines containing ERROR
error_logs = log_rdd.filter(lambda line: "ERROR" in line)

# Filter lines matching multiple conditions
critical_errors = log_rdd.filter(
    lambda line: "ERROR" in line and "database" in line.lower()
)

# Display filtered results
for error in error_logs.take(10):
    print(error)

The classic word count example demonstrates flatMap() and map():

# Split lines into words and count occurrences
word_counts = (
    sc.textFile("data/document.txt")
    .flatMap(lambda line: line.split())
    .map(lambda word: (word.lower(), 1))
    .reduceByKey(lambda a, b: a + b)
)

# Display top 10 most frequent words
top_words = word_counts.takeOrdered(10, key=lambda x: -x[1])
for word, count in top_words:
    print(f"{word}: {count}")

Cleaning text data by removing empty lines and trimming whitespace:

# Remove empty lines and trim whitespace
cleaned_rdd = (
    sc.textFile("data/messy_data.txt")
    .map(lambda line: line.strip())
    .filter(lambda line: len(line) > 0)
)

# More complex cleaning with multiple operations
processed_rdd = (
    sc.textFile("data/input.txt")
    .map(lambda line: line.strip())
    .filter(lambda line: line and not line.startswith("#"))
    .map(lambda line: line.lower())
)

Handling Different File Formats and Edge Cases

PySpark automatically handles compressed files without additional configuration:

# Read gzipped files (automatically decompressed)
gzip_rdd = sc.textFile("data/logs.txt.gz")

# Works with bzip2 as well
bzip_rdd = sc.textFile("data/archive.txt.bz2")

# Mix compressed and uncompressed files
mixed_rdd = sc.textFile("data/*.txt,data/*.gz")

When dealing with malformed data, implement error handling to prevent job failures:

def safe_parse_line(line):
    try:
        # Attempt to parse line (e.g., JSON, CSV)
        parts = line.split(",")
        return {"id": int(parts[0]), "value": float(parts[1])}
    except (ValueError, IndexError) as e:
        # Log error and return None or default value
        return None

# Apply safe parsing and filter out failures
parsed_rdd = (
    sc.textFile("data/potentially_corrupt.csv")
    .map(safe_parse_line)
    .filter(lambda x: x is not None)
)

For files with specific encodings, you may need to handle character encoding explicitly:

# Note: textFile() uses UTF-8 by default
# For other encodings, you might need to use wholeTextFiles() and decode manually
def decode_latin1(file_content):
    filename, content = file_content
    return content.encode('latin1').decode('utf-8')

# Read with custom encoding
custom_encoding_rdd = sc.wholeTextFiles("data/latin1_files/").map(decode_latin1)

Performance Considerations and Best Practices

Caching RDDs is essential when performing multiple operations on the same dataset:

# Load and cache for reuse
data_rdd = sc.textFile("data/large_dataset.txt").cache()

# Multiple operations benefit from caching
error_count = data_rdd.filter(lambda x: "ERROR" in x).count()
warning_count = data_rdd.filter(lambda x: "WARN" in x).count()
info_count = data_rdd.filter(lambda x: "INFO" in x).count()

# Don't forget to unpersist when done
data_rdd.unpersist()

Choose appropriate partition sizes based on your data volume and cluster resources. A good rule of thumb is 2-4 partitions per CPU core:

# For a 10GB file on a 20-core cluster
optimal_partitions = 20 * 3  # 60 partitions
large_file_rdd = sc.textFile("data/10gb_file.txt", minPartitions=optimal_partitions)

Consider when DataFrames might be more appropriate. If your text data has structure (CSV, JSON, logs with consistent format), DataFrames offer better performance through the Catalyst optimizer:

# For structured text data, prefer DataFrames
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()

# More efficient for structured data
df = spark.read.csv("data/structured_data.csv", header=True, inferSchema=True)

However, stick with RDDs when you need:

  • Complex custom parsing logic
  • Line-by-line processing where structure varies
  • Integration with legacy RDD-based code
  • Fine-grained control over partitioning and execution

Always close your SparkContext when finished to release cluster resources:

# Clean up resources
sc.stop()

Understanding how to create and manipulate RDDs from text files gives you the foundation for handling unstructured data in PySpark. While DataFrames have become the standard for structured data, RDDs remain the go-to solution when you need low-level control over text processing operations.

Liked this? There's more.

Every week: one practical technique, explained simply, with code you can use immediately.