s.exceptions.base.PySparkTypeError: [CANNOT_INFER_TYPE_FOR_FIELD] Unable to infe...
🚨 错误信息
File "/opt/spark-3.5.6-bin-hadoop3/python/lib/pyspark.zip/pyspark/sql/types.py", line 1705, in _infer_schema
pyspark.errors.exceptions.base.PySparkTypeError: [CANNOT_INFER_TYPE_FOR_FIELD] Unable to infer the type of the field `avg_optical_flow`.
25/09/04 14:31:11 INFO SparkContext: Invoking stop() from shutdown hook
25/09/04 14:31:11 INFO SparkContext: SparkContext is stopping with exitCode 0.
🤖 AI解决方案
[CANNOT_INFER_TYPE_FOR_FIELD] 无法推断字段 `avg_optical_flow` 的类型。原因分析
Spark 无法自动推断出 avg_optical_flow 字段的数据类型，通常是因为该字段在输入数据中不存在，或者其值不一致导致无法确定单一的数据类型。
解决方案
在创建 Spark DataFrame 时，显式指定 avg_optical_flow 字段的 schema。
如果 avg_optical_flow 字段应该是一个浮点数，可以这样做：
from pyspark.sql.types import StructType, StructField, FloatType
# 假设你有一个 RDD 或 DataFrame
data = [...] # 你的数据
# 示例：如果数据是列表的列表
# data = [[1.0, 2.5], [2.0, 3.1]]
# 定义 schema
schema = StructType([
StructField("other_field", FloatType(), True), # 替换 "other_field" 为你的其他字段名
StructField("avg_optical_flow", FloatType(), True)
])
# 使用 schema 创建 DataFrame
df = spark.createDataFrame(data, schema=schema)如果你的数据是通过读取文件（如 CSV, JSON）获取的，可以在读取时指定 schema：
# 示例：读取 CSV
df = spark.read.csv("your_data.csv", header=True, schema=schema)预防措施
- 明确 Schema 定义：在处理结构化数据时，始终优先定义并使用明确的 Schema，避免依赖 Spark 的自动类型推断。
- 数据质量检查：在数据加载或转换前，进行数据质量检查，确保目标字段存在且数据类型一致。