创建Spark DataFrame.无法推断类型的架构

Question

有人能帮我解决Spark DataFrame的这个问题吗？

当我执行myFloatRDD.toDF()时，我得到一个错误：

TypeError:无法推断类型"float"的架构

我不明白为什么。。。

示例：

myFloatRdd = sc.parallelize([1.0,2.0,3.0])
df = myFloatRdd.toDF()

感谢

Answer 1

SparkSession.createDataFrame在引擎盖下使用，需要Row/tuple/list/dict*或pandas.DataFrame的RDD/list，除非提供了带有DataType的架构。尝试将float转换为元组，如下所示：

myFloatRdd.map(lambda x: (x, )).toDF()

甚至更好：

from pyspark.sql import Row
row = Row("val") # Or some other column name
myFloatRdd.map(row).toDF()

要从标量列表创建DataFrame，您必须直接使用SparkSession.createDataFrame并提供模式***：

from pyspark.sql.types import FloatType
df = spark.createDataFrame([1.0, 2.0, 3.0], FloatType())
df.show()
## +-----+
## |value|
## +-----+
## |  1.0|
## |  2.0|
## |  3.0|
## +-----+

但对于一个简单的范围，最好使用SparkSession.range:

from pyspark.sql.functions import col
spark.range(1, 4).select(col("id").cast("double"))

---

*不再支持。

**Spark SQL还为公开__dict__的Python对象的模式推断提供了有限的支持。

***仅在Spark 2.0或更高版本中支持。

Answer 2

from pyspark.sql.types import IntegerType, Row
mylist = [1, 2, 3, 4, None ]
l = map(lambda x : Row(x), mylist)
# notice the parens after the type name
df=spark.createDataFrame(l,["id"])
df.where(df.id.isNull() == False).show()

基本上，您需要将int初始化到Row((中，然后我们可以使用模式

Answer 3

使用反射推断模式

from pyspark.sql import Row
# spark - sparkSession
sc = spark.sparkContext
# Load a text file and convert each line to a Row.
orders = sc.textFile("/practicedata/orders")
#Split on delimiters
parts = orders.map(lambda l: l.split(","))
#Convert to Row
orders_struct = parts.map(lambda p: Row(order_id=int(p[0]), order_date=p[1], customer_id=p[2], order_status=p[3]))
for i in orders_struct.take(5): print(i)
#convert the RDD to DataFrame
orders_df = spark.createDataFrame(orders_struct)

以编程方式指定模式

from pyspark.sql import Row
# spark - sparkSession
sc = spark.sparkContext
# Load a text file and convert each line to a Row.
orders = sc.textFile("/practicedata/orders")
#Split on delimiters
parts = orders.map(lambda l: l.split(","))
#Convert to tuple
orders_struct = parts.map(lambda p: (p[0], p[1], p[2], p[3].strip()))
#convert the RDD to DataFrame
orders_df = spark.createDataFrame(orders_struct)
# The schema is encoded in a string.
schemaString = "order_id order_date customer_id status"
fields = [StructField(field_name, StringType(), True) for field_name in schemaString.split()]
schema = Struct
ordersDf = spark.createDataFrame(orders_struct, schema)

类型(字段(

Answer 4

from pyspark.sql import Row
myFloatRdd.map(lambda x: Row(x)).toDF()