如何用PySpark将列中特定行转为独立列并加递增索引，及处理大文本文件入DataFrame

Hey there! Let's tackle your PySpark challenges step by step—first loading that 500MB text file into a structured DataFrame, then reshaping your data to extract specific entries into separate columns and adding incrementing indices. I'll keep this beginner-friendly with clear code examples and explanations.

First, since we're dealing with a large file, we want to use PySpark's distributed text reading to avoid memory bottlenecks. We'll start by reading the file line-by-line, then parse each line to separate the main ID (like 1:, 2:) from the raw entry data.

from pyspark.sql import SparkSession
from pyspark.sql.functions import (
    split, explode, regexp_extract, lit, size, sequence, slice,
    monotonically_increasing_id, row_number
)
from pyspark.sql.window import Window

# Initialize your Spark session (adjust configs if needed for large files)
spark = SparkSession.builder \
    .appName("LargeTextToStructuredDF") \
    .config("spark.executor.memory", "4g")  # Optional: Tune based on your cluster resources
    .getOrCreate()

# Read the text file (distributed, no single-node memory overload)
raw_text_df = spark.read.text("path/to/your/large_file.txt")

# Parse each line into main_id and raw entry array
parsed_df = raw_text_df \
    # Extract the main ID (the number before the colon)
    .withColumn("main_id", regexp_extract("value", r"^(\d+):", 1)) \
    # Extract everything after the colon, split into individual elements by commas
    .withColumn("raw_entries", split(regexp_extract("value", r":\s*(.*)", 1), ",\s*"))

Looking at your example, each main ID has multiple comma-separated triples (e.g., 1488844,3,2005-09-06 822109). We need to split these triples into separate rows, then break each triple into its own columns.

# Calculate how many triples exist per row (each triple has 3 elements)
parsed_df = parsed_df.withColumn("num_triples", (size("raw_entries") / 3).cast("int"))

# Generate indices to slice the raw_entries array into individual triples
parsed_df = parsed_df.withColumn("triple_indices", sequence(lit(0), col("num_triples") - 1))

# Explode the indices to create one row per triple
exploded_triples_df = parsed_df \
    .withColumn("triple_idx", explode("triple_indices")) \
    # Slice the raw_entries array to get each 3-element triple
    .withColumn("triple", slice(col("raw_entries"), col("triple_idx") * 3 + 1, 3))

# Split the triple array into distinct columns (adjust names based on your actual data meaning)
structured_df = exploded_triples_df \
    .withColumn("user_id", col("triple")[0]) \
    .withColumn("rating", col("triple")[1].cast("int"))  # Cast to int if it's a numeric value
    .withColumn("date_detail", col("triple")[2]) \
    # Keep only the columns we need
    .select("main_id", "user_id", "rating", "date_detail")

You have two common options for indices, depending on your needs:

Option 1: Global Incrementing Index (Unique Across All Rows) #

Use monotonically_increasing_id() for a globally unique, incrementing index (note: this is not strictly consecutive but guaranteed to be increasing):

structured_df = structured_df.withColumn("global_index", monotonically_increasing_id())

Option 2: Group-Wise Incrementing Index (Per Main ID) #

If you want an index that resets for each main_id (e.g., 1,2,3 for entries under main_id=1, then 1,2,3 for main_id=2), use a window function:

# Define a window partitioned by main_id, ordered by date_detail (adjust sort column as needed)
window_spec = Window.partitionBy("main_id").orderBy("date_detail")

# Add the row number as the group index
structured_df = structured_df.withColumn("group_index", row_number().over(window_spec))

You can check the output with:

structured_df.show(10, truncate=False)

This approach scales well for large files because all operations are distributed across your Spark cluster—no need to load the entire 500MB file into a single machine's memory.

内容的提问来源于stack exchange，提问作者user1977981