使用PySpark将纯文本转CSV或DataFrame并创建Hive表的方法

1 无需转换源文件直接创建Hive表的可行性 #

可以实现，核心是使用Hive的正则SerDe类org.apache.hadoop.hive.serde2.RegexSerDe直接解析原始文本的标签格式，无需提前预处理文件。
该方案仅适合临时查询、数据量较小的场景，生产环境如果无效行占比过高，会增加查询时的过滤开销，建议提前做数据清洗。
参考建表语句：

CREATE EXTERNAL TABLE IF NOT EXISTS raw_tag_data (
    name string,
    code string,
    value string
)
ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.RegexSerDe'
WITH SERDEPROPERTIES (
    "input.regex" = "<__name__>(.*?)<__code__>(.*?)<__value__>(.*?)",
    "input.regex.case.insensitive" = "false"
)
STORED AS TEXTFILE
LOCATION '/your/hdfs/path/to/raw/files';

查询时可增加WHERE name IS NOT NULL条件过滤掉不符合正则匹配的无效行。 #

2 PySpark转换为结构化数据的实现方案 #

实现逻辑 #

• 读取原始文本文件，每行作为一个单独的字符串字段
• 通过正则提取三个标签对应的字段值
• 过滤无效行（未匹配到三个标签的行直接丢弃）
• 输出为带表头的CSV文件，或者直接注册为临时表供后续使用

参考代码 #

from pyspark.sql import SparkSession
from pyspark.sql.functions import regexp_extract, col

# 初始化SparkSession
spark = SparkSession.builder.appName("tag_data_parse").getOrCreate()

# 1. 读取原始文本文件
raw_df = spark.read.text("/your/input/path/raw_files/*.txt")

# 2. 正则提取三个字段，分组1对应name，分组2对应code，分组3对应value
parse_df = raw_df.select(
    regexp_extract(col("value"), r"<__name__>(.*?)<__code__>(.*?)<__value__>(.*?)", 1).alias("name"),
    regexp_extract(col("value"), r"<__name__>(.*?)<__code__>(.*?)<__value__>(.*?)", 2).alias("code"),
    regexp_extract(col("value"), r"<__name__>(.*?)<__code__>(.*?)<__value__>(.*?)", 3).alias("value")
)

# 3. 过滤无效行：丢弃三个字段全为空的未匹配行
clean_df = parse_df.filter("name != '' OR code != '' OR value != ''")

# 可选：输出为带表头的CSV文件，合并为单个文件（数据量大时可删除coalesce(1)）
clean_df.coalesce(1).write.option("header", "true").csv("/your/output/path/csv_result")

# 可选：直接转为临时表供Hive/Spark SQL查询
clean_df.createOrReplaceTempView("v_clean_tag_data")

如果标签内容中间存在换行的情况，读取文件时增加option("lineSep", "<__name__>")参数按标签拆分记录即可。

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