PySpark非正则方式提取日志键值对并转换数据格式求助

我有一个仅包含_raw列的CSV文件，样例数据如下：

_raw
2022-11-22 23:31:23,408               customer=customer1 instance=instance1 user=userA page="page1" subpage="subpage1" behavior=behavior1 ct=ctA id=id3
2022-10-24 20:10:23,408               customer=customer2 instance=instance2 user=userB page="page2" subpage="subpage2" behavior=behavior2 ct=ctB id=id2
...

希望通过PySpark将其转换为如下结构化格式：

_time, customer, instance, user, page, subpage, behavior
2022-11-22 23:31:23,408, customer1, instance1, userA, page1, subpage1, behavior1
2022-10-24 20:10:23,408, customer2, instance2, userB, page2, subpage2, behavior2
...

我已尝试使用正则表达式的regexp_extract方法实现，但提取多列时操作繁琐，请问是否有更简便的实现方式？附尝试的正则代码：

df.withColumn("customer", regexp_extract(col('_raw'), 'customer=(\S+)', 1))\
  .withColumn("instance", regexp_extract(col('_raw'), 'instance=(\S+)', 1 ))\
  .withColumn(...)
  ...

这里提供三种更高效的实现方式，避免重复调用regexp_extract：

方法一：拆分字段后转换为键值对映射 #

先拆分每行的时间部分和键值对部分，将键值对转为Map类型后直接提取目标字段：

from pyspark.sql import functions as F
from pyspark.sql.types import MapType, StringType

def parse_raw_line(line):
    parts = line.split()
    time_part = parts[0]
    kv_parts = parts[1:]
    kv_map = {}
    for kv in kv_parts:
        key, value = kv.split('=', 1)
        # 去除字段值的引号
        if value.startswith('"') and value.endswith('"'):
            value = value[1:-1]
        kv_map[key] = value
    return (time_part, kv_map)

# 注册自定义UDF并指定返回类型
parse_udf = F.udf(parse_raw_line, StringType() + MapType(StringType(), StringType()))

result_df = df.withColumn("parsed", parse_udf(F.col("_raw")))\
    .select(
        F.col("parsed._1").alias("_time"),
        F.col("parsed._2.customer").alias("customer"),
        F.col("parsed._2.instance").alias("instance"),
        F.col("parsed._2.user").alias("user"),
        F.col("parsed._2.page").alias("page"),
        F.col("parsed._2.subpage").alias("subpage"),
        F.col("parsed._2.behavior").alias("behavior")
    )

方法二：单正则一次性捕获所有字段 #

编写匹配整行的正则表达式，一次捕获所有需要的字段，减少正则调用次数：

from pyspark.sql import functions as F

# 匹配整行的正则，依次捕获时间和各字段值
pattern = r'^(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d{3})\s+customer=(\S+)\s+instance=(\S+)\s+user=(\S+)\s+page="?([^"]+)"?\s+subpage="?([^"]+)"?\s+behavior=(\S+)'

result_df = df.select(
    F.regexp_extract(F.col("_raw"), pattern, 1).alias("_time"),
    F.regexp_extract(F.col("_raw"), pattern, 2).alias("customer"),
    F.regexp_extract(F.col("_raw"), pattern, 3).alias("instance"),
    F.regexp_extract(F.col("_raw"), pattern, 4).alias("user"),
    F.regexp_extract(F.col("_raw"), pattern, 5).alias("page"),
    F.regexp_extract(F.col("_raw"), pattern, 6).alias("subpage"),
    F.regexp_extract(F.col("_raw"), pattern, 7).alias("behavior")
)

方法三：用内置拆分+透视函数处理 #

适合字段不固定的场景，通过拆分、 explode、 pivot完成结构化转换：

from pyspark.sql import functions as F

result_df = df.withColumn("split_line", F.split(F.col("_raw"), "\s{2,}"))\
    .withColumn("_time", F.col("split_line")[0])\
    .withColumn("kv_pairs", F.split(F.col("split_line")[1], "\s"))\
    .withColumn("kv", F.explode(F.col("kv_pairs")))\
    .withColumn("key", F.split(F.col("kv"), "=")[0])\
    .withColumn("value", F.regexp_replace(F.split(F.col("kv"), "=")[1], '"', ''))\
    .groupBy("_time")\
    .pivot("key")\
    .agg(F.first("value"))\
    .select("_time", "customer", "instance", "user", "page", "subpage", "behavior")

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