为从文件加载的Spark DataFrame添加Schema时出现类型转换错误求助

The error pops up because when you read the CSV file without specifying a schema, Spark automatically infers all columns as StringType. When you try to create a new DataFrame using spark.createDataFrame(tableDF, schemaTd), Spark doesn’t do implicit type conversion—it expects the input data to already match the schema’s data types. Since your tableDF has all String values, trying to interpret them as Integers throws the "String is not a valid external type for schema of int" error.

Here are the best ways to fix this:

Solution 1: Read the CSV with Your Schema Directly (Recommended) #

This is the most efficient approach—tell Spark the schema upfront so it parses the columns into the correct types during reading:

import org.apache.spark.sql.types.{StringType, StructField, StructType, IntegerType}

// Define your schema
val schemaTd = StructType(List(
  StructField("time_id", IntegerType),
  StructField("week", IntegerType),
  StructField("month", IntegerType),
  StructField("calendar", StringType)
))

// Read CSV with the schema
val result = spark.read
  .option("delimiter", ",")
  .schema(schemaTd)
  .csv("/Volumes/Data/ap/click/test.csv")

// Now this will work
result.show()

Solution 2: Cast Columns Explicitly After Reading #

If you already have the raw String-based DataFrame, cast each column to the desired type using the DataFrame API:

import org.apache.spark.sql.functions.col
import org.apache.spark.sql.types.IntegerType

val tableDF = spark.read.option("delimiter", ",").csv("/Volumes/Data/ap/click/test.csv")

val result = tableDF.select(
  col("_c0").cast(IntegerType).alias("time_id"),
  col("_c1").cast(IntegerType).alias("week"),
  col("_c2").cast(IntegerType).alias("month"),
  col("_c3").alias("calendar")
)

result.show()

Handling Invalid Data #

If your CSV might contain non-integer values (which would break the cast), use try_cast (Spark 2.3+) instead—it returns null for invalid conversions instead of crashing your job:

col("_c0").try_cast(IntegerType).alias("time_id")

Solution 3: Convert Rows Before Creating DataFrame (Less Efficient) #

You can convert the RDD of rows to match the schema types first, but this uses the older RDD API and is less optimal than the above methods:

val convertedRows = tableDF.rdd.map(row => {
  Row(
    row.getString(0).toInt,
    row.getString(1).toInt,
    row.getString(2).toInt,
    row.getString(3)
  )
})

val result = spark.createDataFrame(convertedRows, schemaTd)
result.show()

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