如何使用Python/PySpark移除CSV文件的前N行与后N行

嘿，我来帮你搞定这个问题！针对CSV文件前后带有多余元数据行的情况，不管是处理小文件用纯Python，还是大数据场景用PySpark，都有很直接的解决方案，下面给你详细拆解：

一、纯Python处理小文件 #

如果你的CSV文件不大，用纯Python就能轻松搞定，核心思路是读取所有行后切片去掉不需要的部分，再写回文件：

# 读取原始CSV文件
with open("your_source.csv", "r", encoding="utf-8") as infile:
    all_lines = infile.readlines()

# 移除前10行元数据和最后2行元数据
# 切片语法：[起始索引:结束索引]，-2表示取到倒数第2行之前的内容
cleaned_lines = all_lines[10:-2]

# 将清理后的内容写入新文件
with open("cleaned_output.csv", "w", encoding="utf-8") as outfile:
    outfile.writelines(cleaned_lines)

如果之后需要用pandas解析处理后的文件，记得指定分隔符是分号：

import pandas as pd
df = pd.read_csv("cleaned_output.csv", delimiter=";")

二、PySpark处理大文件 #

如果是大数据场景，PySpark的分布式处理更高效，这里提供两种常用方法：

方法1：通过行号过滤（适合固定行数的元数据） #

这种方法精准控制要移除的前后行数，步骤如下：

from pyspark.sql import SparkSession
from pyspark.sql.window import Window
from pyspark.sql.functions import row_number, split

# 初始化Spark会话
spark = SparkSession.builder.appName("CleanCSVMetadata").getOrCreate()

# 先以文本格式读取整个CSV（因为前10行是元数据，不能直接按结构化表格读取）
raw_text_df = spark.read.text("your_large_file.csv")

# 获取文件总行数
total_rows = raw_text_df.count()

# 给每一行添加行号
window_spec = Window.orderBy("value")
df_with_row_num = raw_text_df.withColumn("row_num", row_number().over(window_spec))

# 过滤掉前10行和最后2行
filtered_df = df_with_row_num.filter(
    (df_with_row_num.row_num > 10) & (df_with_row_num.row_num <= total_rows - 2)
)

# 提取表头（第11行，也就是过滤后的第一行）
header = filtered_df.filter(filtered_df.row_num == 11).select("value").first()[0].split(";")

# 提取数据行并转换为结构化DataFrame
data_df = filtered_df.filter(filtered_df.row_num > 11).select(
    *[split("value", ";")[i].alias(header[i]) for i in range(len(header))]
)

# 保存清理后的结构化CSV
data_df.write.csv("spark_cleaned_csv", header=True, sep=";", mode="overwrite")

方法2：通过内容过滤（如果元数据行有固定标识） #

如果你的元数据行有明显特征（比如最后两行包含"END_REPORT"之类的关键词），可以直接按内容过滤，更灵活：

raw_text_df = spark.read.text("your_large_file.csv")

# 过滤掉前10行，同时过滤掉包含特定关键词的最后两行
cleaned_text_df = raw_text_df.filter(
    (row_number().over(Window.orderBy("value")) > 10) & 
    (~raw_text_df.value.contains("END_REPORT"))
)

备注：内容来源于stack exchange，提问作者Anjikya