如何在PySpark中为半结构化文本文件定义指定数据类型的Schema

问题背景 #

现有半结构化文本数据，列分隔规则为：

• 第1、2列：空格分隔
• 第2、3列：制表符（\t）分隔
• 第3、4列：逗号（,）分隔

需要将各列转换为指定类型：

• 第1列：int（order_id）
• 第2列：timestamp（date）
• 第3列：int（customer_id）
• 第4列：string（status）

用户通过正则表达式提取出所有列，但结果均为string类型；尝试用spark.createDataFrame指定预定义Schema时失败，需找到正确的类型转换方法。

用户原尝试代码：

from pyspark import SparkConf
from pyspark.sql import SparkSession
from pyspark.sql.functions import regexp_extract
from pyspark.sql.types import IntegerType, StructField, StructType, StringType, TimestampType
 
my_conf = SparkConf()
my_conf.set("spark.app.name", "my first application")
my_conf.set("spark.master","local[*]")
spark = SparkSession.builder.config(conf=my_conf).getOrCreate()

schema1 = StructType([StructField("order_id", IntegerType(),True),
StructField("date", TimestampType(),True),
StructField("customer_id", IntegerType(),True),
StructField("status", StringType(),True)])

myregex = r'^(\S+) (\S+)\t(\S+)\,(\S+)'
lines_df = spark.read.format("text").option("path","C:/Users/Lenovo/Desktop/week11/week 11 datasets/orders_new.csv").load()
 
final_df =lines_df.select(regexp_extract('value',myregex,1).alias("order_id"),
regexp_extract('value',myregex,2).alias("date"),
regexp_extract('value',myregex,3).alias("customer_id"),
regexp_extract('value',myregex,4).alias("status"))

# 尝试指定Schema时出错
df=spark.createDataFrame(final_df.rdd,schema1)
final_df.show()

错误原因 #

regexp_extract返回的结果始终是string类型，直接将final_df.rdd传入createDataFrame并指定Schema时，Spark会尝试自动转换类型，但可能因时间格式不匹配、空值/非法值存在等原因转换失败；此外这种方法绕开了DataFrame的优化，效率较低。

正确解决方案 #

方案1：DataFrame层面直接转换类型（推荐） #

利用cast函数转换数值类型，用to_timestamp处理时间字段（可指定格式确保转换成功）：

from pyspark import SparkConf
from pyspark.sql import SparkSession
from pyspark.sql.functions import regexp_extract, to_timestamp
from pyspark.sql.types import IntegerType, StringType, TimestampType
 
my_conf = SparkConf()
my_conf.set("spark.app.name", "my first application")
my_conf.set("spark.master","local[*]")
spark = SparkSession.builder.config(conf=my_conf).getOrCreate()

myregex = r'^(\S+) (\S+)\t(\S+)\,(\S+)'
lines_df = spark.read.format("text").option("path","C:/Users/Lenovo/Desktop/week11/week 11 datasets/orders_new.csv").load()

# 提取后直接转换类型
final_df = lines_df.select(
    regexp_extract('value', myregex, 1).cast(IntegerType()).alias("order_id"),
    to_timestamp(regexp_extract('value', myregex, 2)).alias("date"),  # 若时间格式特殊，可指定格式：to_timestamp(col, "yyyy-MM-dd HH:mm:ss")
    regexp_extract('value', myregex, 3).cast(IntegerType()).alias("customer_id"),
    regexp_extract('value', myregex, 4).alias("status")
)

# 验证Schema和数据
final_df.printSchema()
final_df.show()

方案2：通过RDD转换后指定Schema #

若必须使用RDD方式，需先将每个元素的string值手动转换为目标类型，再传入createDataFrame：

from pyspark import SparkConf
from pyspark.sql import SparkSession
from pyspark.sql.functions import regexp_extract
from pyspark.sql.types import IntegerType, StructField, StructType, StringType, TimestampType
from datetime import datetime
 
my_conf = SparkConf()
my_conf.set("spark.app.name", "my first application")
my_conf.set("spark.master","local[*]")
spark = SparkSession.builder.config(conf=my_conf).getOrCreate()

schema1 = StructType([
    StructField("order_id", IntegerType(), True),
    StructField("date", TimestampType(), True),
    StructField("customer_id", IntegerType(), True),
    StructField("status", StringType(), True)
])

myregex = r'^(\S+) (\S+)\t(\S+)\,(\S+)'
lines_df = spark.read.format("text").option("path","C:/Users/Lenovo/Desktop/week11/week 11 datasets/orders_new.csv").load()

# 提取字符串列
temp_df = lines_df.select(
    regexp_extract('value', myregex, 1).alias("order_id"),
    regexp_extract('value', myregex, 2).alias("date"),
    regexp_extract('value', myregex, 3).alias("customer_id"),
    regexp_extract('value', myregex, 4).alias("status")
)

# 将RDD的每个元素转换为对应类型的tuple
def convert_row(row):
    return (
        int(row.order_id) if row.order_id else None,
        datetime.strptime(row.date, "%Y-%m-%d %H:%M:%S") if row.date else None,  # 替换为你的时间格式
        int(row.customer_id) if row.customer_id else None,
        row.status
    )

converted_rdd = temp_df.rdd.map(convert_row)
df = spark.createDataFrame(converted_rdd, schema1)

# 验证
df.printSchema()
df.show()

注意事项 #

• 使用to_timestamp时，若时间格式不是Spark默认的格式（yyyy-MM-dd HH:mm:ss），必须指定格式参数，例如to_timestamp(col, "dd/MM/yyyy")
• 若数据中存在空值或非法值（比如非数字的order_id），转换时会返回null，可结合coalesce或when函数处理异常值

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