PySpark中使用spark.write.saveAsTable从CSV创建表时如何指定列类型

Spark默认读取CSV时会将所有列解析为string类型，想要直接指定正确类型，只需在读取阶段传入自定义schema即可，以下是三种可行方案：

方法1：显式定义StructType Schema #

这是最精准可控的方式，直接为每一列指定对应的数据类型：

from pyspark.sql import DataFrame
from pyspark.sql.types import StructType, StructField, DateType, StringType, DoubleType

import_path = f"{st_raw}/data.csv"

# 匹配数据集的自定义schema
custom_schema = StructType([
    StructField("Date", DateType(), nullable=True),
    StructField("Location", StringType(), nullable=True),
    StructField("Country", StringType(), nullable=True),
    StructField("Temperature", DoubleType(), nullable=True),
    StructField("CO2 Emissions", DoubleType(), nullable=True),
    StructField("Sea Level Rise", DoubleType(), nullable=True),
    StructField("Precipitation", DoubleType(), nullable=True),
    StructField("Humidity", DoubleType(), nullable=True),
    StructField("Wind Speed", DoubleType(), nullable=True)
])

# 读取CSV时指定schema
sparkDF = spark.read.csv(import_path, header=True, schema=custom_schema)

spark.sql(f"CREATE SCHEMA IF NOT EXISTS {catalog}.{schema}")

tablename = f"{catalog}.{schema}.{table}"
sparkDF.write.saveAsTable(tablename)

assert spark.table(tablename).count() > 0
display(spark.table(tablename))

方法2：使用DDL字符串定义Schema #

如果觉得StructType写法繁琐，可用DDL格式字符串定义，语法更简洁：

from pyspark.sql import DataFrame

import_path = f"{st_raw}/data.csv"

# DDL格式的schema定义，含空格的列名需用反引号包裹
ddl_schema = """
    Date DATE,
    Location STRING,
    Country STRING,
    Temperature DOUBLE,
    `CO2 Emissions` DOUBLE,
    `Sea Level Rise` DOUBLE,
    Precipitation DOUBLE,
    Humidity DOUBLE,
    `Wind Speed` DOUBLE
"""

# 读取CSV时传入schema参数
sparkDF = spark.read.csv(import_path, header=True, schema=ddl_schema)

spark.sql(f"CREATE SCHEMA IF NOT EXISTS {catalog}.{schema}")

tablename = f"{catalog}.{schema}.{table}"
sparkDF.write.saveAsTable(tablename)

assert spark.table(tablename).count() > 0
display(spark.table(tablename))

方法3：使用inferSchema自动推断（慎用） #

若数据集规模较小，可开启inferSchema让Spark自动推断类型，但该方式在大数据集上会降低读取速度，且部分特殊格式的字段可能推断不准确：

from pyspark.sql import DataFrame

import_path = f"{st_raw}/data.csv"

# 开启自动类型推断
sparkDF = spark.read.csv(import_path, header=True, inferSchema=True)

spark.sql(f"CREATE SCHEMA IF NOT EXISTS {catalog}.{schema}")

tablename = f"{catalog}.{schema}.{table}"
sparkDF.write.saveAsTable(tablename)

assert spark.table(tablename).count() > 0
display(spark.table(tablename))

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