如何用PySpark 2.0+读取含多分隔符的CSV文件？

嘿，这个场景我太熟悉了！PySpark的CSV读取器确实没法直接处理这种多分隔符、格式偏日志风格的数据——毕竟它的delimiter只能指定单个字符，也不支持类Grok的正则模式。不过咱们可以换个思路，用文本读取+正则解析的方式搞定，给你两个实用的方案：

方法一：纯Spark SQL正则函数处理（性能优先） #

这种方法不用写自定义函数，完全依赖Spark内置的正则函数，性能更好，适合处理大规模数据。

步骤1：按文本读取整个文件 #

首先把文件当成纯文本读取，保留每一行的原始内容：

df = spark.read.text("path/to/your/log_file.log")

步骤2：拆分出独立的记录 #

看你提供的示例，每条记录都是以<31>开头的，甚至可能一行里包含多条记录。我们可以用regexp_extract_all匹配所有符合格式的记录，再用explode把它们拆成单独的行：

from pyspark.sql.functions import regexp_extract_all, explode

# 匹配所有以<31>开头，直到下一个<31>或行尾的内容
df = df.withColumn("records", regexp_extract_all("value", r"<31>.*?(?=<31>|$)", 0))
# 把数组拆成单独的记录行
df = df.select(explode("records").alias("raw_record"))

步骤3：提取各个字段 #

现在每条raw_record的格式是固定的，我们可以用regexp_extract精准匹配每个字段——包括结尾带空格的文本字符串：

from pyspark.sql.functions import regexp_extract

df = df \
    .withColumn("timestamp", regexp_extract("raw_record", r"^(\w{3} \d{2} \d{2}:\d{2}:\d{2})", 1)) \
    .withColumn("device_name", regexp_extract("raw_record", r"^\w{3} \d{2} \d{2}:\d{2}:\d{2} (\S+)", 1)) \
    .withColumn("mac_address", regexp_extract("raw_record", r"^\w{3} \d{2} \d{2}:\d{2}:\d{2} \S+ (\S+)", 1)) \
    .withColumn("ip_address", regexp_extract("raw_record", r"\((\d+\.\d+\.\d+\.\d+)\)", 1)) \
    .withColumn("message", regexp_extract("raw_record", r": (.+)$", 1))

# 查看结果
df.show(truncate=False)

这里最后一个message字段用(.+)$匹配，会完整保留结尾带空格的文本内容，比如idle timeout <600> from RADIUS。

方法二：自定义UDF（灵活处理复杂/异常格式） #

如果你的日志格式有一些不固定的异常情况，纯正则函数处理起来麻烦，可以用Python自定义UDF来解析，逻辑更灵活：

步骤1：定义解析函数和Schema #

import re
from pyspark.sql.functions import udf
from pyspark.sql.types import StructType, StructField, StringType

# 定义输出的Schema，对应每个字段
record_schema = StructType([
    StructField("timestamp", StringType()),
    StructField("device_name", StringType()),
    StructField("mac_address", StringType()),
    StructField("ip_address", StringType()),
    StructField("message", StringType())
])

# 编写解析逻辑，处理每条记录
def parse_log_record(record):
    # 匹配固定格式的正则模式
    pattern = r"^(\w{3} \d{2} \d{2}:\d{2}:\d{2}) (\S+) (\S+) \((\d+\.\d+\.\d+\.\d+)\): (.+)$"
    match_result = re.match(pattern, record)
    if match_result:
        return (
            match_result.group(1),
            match_result.group(2),
            match_result.group(3),
            match_result.group(4),
            match_result.group(5)
        )
    # 处理格式异常的记录，返回空值
    else:
        return (None, None, None, None, None)

# 把函数转换成Spark UDF
parse_udf = udf(parse_log_record, record_schema)

步骤2：应用UDF解析记录 #

# 先按方法一的步骤拆分出raw_record，再应用UDF
df_parsed = df.select(parse_udf("raw_record").alias("parsed_data"))
# 展开嵌套的结构体字段
df_parsed = df_parsed.select("parsed_data.*")

df_parsed.show(truncate=False)

关键说明 #

• PySpark的CSV读取器确实不支持多分隔符或正则分隔符，所以必须先按文本读取，再做后续解析。
• 优先用方法一的内置正则函数，因为UDF会把数据拉到Python端处理，性能比Spark原生函数差，适合小数据或复杂格式场景。

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