在将Pandas数据帧转换为Spark数据帧时，是否可以将Float转换为Long？

问题描述

我有以下两个方案共享的前奏代码：

from pyspark.sql import SparkSession
from pyspark.sql.types import *
import pyspark.sql.functions as F
    
import pandas as pd
import numpy as np
    
spark = SparkSession.builder.getOrCreate()

df = pd.DataFrame({"col1": [1, 2, 3], "col2": [22.0, 88.0, np.nan]})

现在，我想将df转换为pyspark数据帧(sdf)。在创建sdf期间，当我尝试通过架构将"col2"隐式转换为LongType时失败：

schema = StructType([StructField("col1", LongType()), StructField("col2", LongType())])
sdf = spark.createDataFrame(df[schema.fieldNames()], schema=schema)

错误：

TypeError：字段Col2：LongType无法接受类型中的对象22.0 <；class‘Float’>；

但如果我运行以下代码段，它就能正常工作：

schema_2 = StructType(
    [StructField("col1", LongType()), StructField("col2", FloatType())]
)
sdf = spark.createDataFrame(df[schema.fieldNames()], schema=schema_2)
cast_sdf = sdf.withColumn("col2", F.col("col2").cast(LongType()))
cast_sdf.show()

输出：

+----+----+                                                                     
|col1|col2|
+----+----+
|   1|  22|
|   2|  88|
|   3|   0|
+----+----+

---

解决方案

将我的评论转换为答案。

这实际上就是Spark处理模式的方式。它并不特定于 pandas 数据帧被转换为星火数据帧。将createDataframe方法与元组列表一起使用时，您将收到相同的错误：

import numpy as np

schema = StructType([StructField("col1", LongType()), StructField("col2", LongType())])
df = spark.createDataFrame([(1, 22.0), (2, 88.0), (3, np.nan)], schema)

# TypeError: field col2: LongType can not accept object 22.0 in type <class 'float'>

这也是像CSV这样的数据源在传递模式时的行为(尽管当读取CSV时，它不会在模式PERMISSIVE下失败，但值被加载为空)。因为模式不自动转换类型，所以它只告诉Spark行中的每一列应该有哪种数据类型。

因此，在使用架构时，您必须传递与指定类型匹配的数据，或者使用不失败的StringType，然后使用显式强制转换将列转换为所需的类型。

schema = StructType([StructField("col1", LongType()), StructField("col2", StringType())])

df = spark.createDataFrame([(1, 22.0), (2, 88.0), (3, np.nan)], schema)

df = df.withColumn("col2", F.col("col2").cast("long"))
df.show()

#+----+----+
#|col1|col2|
#+----+----+
#|   1|  22|
#|   2|  88|
#|   3|null|
#+----+----+