Spark Python如何从日志CSV数据生成指定结构的DataFrame

需求说明 #

读取日志文件存储路径，从日志中提取CSV文本表数据，输出包含CSV原表字段+序列号+时间戳的DataFrame，最终可输出表数据的JSON格式，具体要求如下：

• 首先从数据库读取serial_no、test_date、s3_path三个字段的元数据
• s3_path字段存储对应CSV日志文件的S3路径
• 输出要求为包含CSV表全部字段、主键、时间戳的结构化DataFrame

现有实现进度 #

已通过如下伪代码从Redshift读取元数据得到初始DataFrame：

df = sparkSession.read \
            .format("com.databricks.spark.redshift") \
            .option("url",
                    "some url with id{}&password={}".format(
                        redshift_user, redshift_pass)) \
            .option("query", query) \
            .option("tempdir", s3_redshift_temp_dir) \
            .option("forward_spark_s3_credentials", True)
df = df_context.load()

初始DataFrame结构示例：

+-------------+-------------------+--------------------+
|serial_number|          test_date|             s3_path|
+-------------+-------------------+--------------------+
|     A0123456|2019-07-10 04:11:52|s3://test-bucket-...|
|     A0123456|2019-07-24 23:48:03|s3://test-bucket-...|
+-------------+-------------------+--------------------+

由于无法将SparkContext传递到worker节点，已通过boto3实现读取S3文件的UDF，注册后调用得到包含CSV字符串字段的中间DataFrame：

spark.udf.register("read_s3_file", read_s3_file)
df_with_string_csv = df.withColumn('table_csv_data', read_s3_file(drive_event_tab.s3_path))

中间DataFrame示例如下，其中table_csv_data字段存储完整的带表头CSV文本：

+-------------+-------------------+--------------------+----------------------+
|serial_number|          test_date|             s3_path|    table_csv_data    |
+-------------+-------------------+--------------------+----------------------+
|     1050D1B0|2019-05-07 15:41:11|s3://test-bucket-...|timestamp,partition,offset,key,value...|
+-------------+-------------------+--------------------+----------------------+

table_csv_data字段内容示例：

timestamp,partition,offset,key,value
1625218801350,97,33009,2CKXTKAT_20210701193302_6400_UCMP,458969040
1625218801349,41,33018,3FGW9S6T_20210701193210_6400_UCMP,17569160

实现方案（得到目标结构化DataFrame） #

依赖导入 #

from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, LongType, StringType

步骤1：定义CSV字段Schema（可选，推荐显式指定避免类型推断错误） #

csv_schema = StructType([
    StructField("timestamp", LongType(), nullable=True),
    StructField("partition", LongType(), nullable=True),
    StructField("offset", LongType(), nullable=True),
    StructField("key", StringType(), nullable=True),
    StructField("value", LongType(), nullable=True)
])

步骤2：拆分CSV字符串并行转列、解析字段 #

Spark 3.0及以上版本（推荐使用from_csv内置函数）

# 按换行符拆分CSV文本为行数组
df_split = df_with_string_csv.withColumn("csv_lines", F.split(F.col("table_csv_data"), "\n"))
# 过滤空行、表头行后炸开为多行
df_explode = df_split.withColumn("csv_line", F.explode(F.expr("filter(csv_lines, x -> x is not null and trim(x) != '' and not x starts with 'timestamp,')")))
# 按Schema解析CSV行
df_parsed = df_explode.withColumn("csv_fields", F.from_csv(F.col("csv_line"), schema=csv_schema, options={"sep": ","}))
# 提取所需字段得到最终DataFrame
final_df = df_parsed.select(
    "serial_number",
    "test_date",
    F.col("csv_fields.*")
)

Spark 2.x版本兼容方案（手动拆分字段）

# 按换行符拆分CSV文本为行数组
df_split = df_with_string_csv.withColumn("csv_lines", F.split(F.col("table_csv_data"), "\n"))
# 过滤空行、表头行后炸开为多行
df_explode = df_split.withColumn("csv_line", F.explode(F.expr("filter(csv_lines, x -> x is not null and trim(x) != '' and not x like 'timestamp,%')")))
# 按逗号拆分每行字段，强转类型后取别名
df_explode = df_explode.withColumn("fields", F.split(F.col("csv_line"), ","))
final_df = df_explode.select(
    "serial_number",
    "test_date",
    F.col("fields")[0].cast(LongType()).alias("timestamp"),
    F.col("fields")[1].cast(LongType()).alias("partition"),
    F.col("fields")[2].cast(LongType()).alias("offset"),
    F.col("fields")[3].alias("key"),
    F.col("fields")[4].cast(LongType()).alias("value")
)

步骤3：输出JSON格式（可选） #

final_df.write.json("你的存储路径", mode="overwrite")

最终得到的DataFrame结构与需求完全一致。

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