我正在从 RDD 创建一个 DataFrame,其中一个值是一个 date
。我不知道如何在架构中指定 DateType()
。
让我来说明手头的问题-
我们可以将 date
加载到 DataFrame 中的一种方法是首先将其指定为字符串,然后使用 to_date() 将其转换为正确的 date
功能。
from pyspark.sql.types import Row, StructType, StructField, StringType, IntegerType, DateType
from pyspark.sql.functions import col, to_date
values=sc.parallelize([(3,'2012-02-02'),(5,'2018-08-08')])
rdd= values.map(lambda t: Row(A=t[0],date=t[1]))
# Importing date as String in Schema
schema = StructType([StructField('A', IntegerType(), True), StructField('date', StringType(), True)])
df = sqlContext.createDataFrame(rdd, schema)
# Finally converting the string into date using to_date() function.
df = df.withColumn('date',to_date(col('date'), 'yyyy-MM-dd'))
df.show()
+---+----------+
| A| date|
+---+----------+
| 3|2012-02-02|
| 5|2018-08-08|
+---+----------+
df.printSchema()
root
|-- A: integer (nullable = true)
|-- date: date (nullable = true)
有没有一种方法,我们可以在 schema
中使用 DateType()
并避免将 string
转换为 date
明确?
像这样的——
values=sc.parallelize([(3,'2012-02-02'),(5,'2018-08-08')])
rdd= values.map(lambda t: Row(A=t[0],date=t[1]))
# Somewhere we would need to specify date format 'yyyy-MM-dd' too, don't know where though.
schema = StructType([StructField('A', DateType(), True), StructField('date', DateType(), True)])
更新:根据@user10465355的建议,以下代码有效 -
import datetime
schema = StructType([
StructField('A', IntegerType(), True),
StructField('date', DateType(), True)
])
rdd= values.map(lambda t: Row(A=t[0],date=datetime.datetime.strptime(t[1], "%Y-%m-%d")))
df = sqlContext.createDataFrame(rdd, schema)
df.show()
+---+----------+
| A| date|
+---+----------+
| 3|2012-02-02|
| 5|2018-08-08|
+---+----------+
df.printSchema()
root
|-- A: integer (nullable = true)
|-- date: date (nullable = true)
最佳答案
长话短说,与外部对象的 RDD
一起使用的模式不应以这种方式使用 - 声明的类型应该反射(reflect)数据的实际状态,而不是所需的状态。
换句话说,允许:
schema = StructType([
StructField('A', IntegerType(), True),
StructField('date', DateType(), True)
])
date
字段对应的数据should use datetime.date
.因此,例如您的 RDD[Tuple[int, str]]
:
import datetime
spark.createDataFrame(
# Since values from the question are just two element tuples
# we can use mapValues to transform the "value"
# but in general case you'll need map
values.mapValues(datetime.date.fromisoformat),
schema
)
最接近所需行为的方法是使用 dicts
使用 JSON 阅读器转换数据 (RDD[Row]
)
from pyspark.sql import Row
spark.read.schema(schema).json(rdd.map(Row.asDict))
或更好的显式 JSON 转储:
import json
spark.read.schema(schema).json(rdd.map(Row.asDict).map(json.dumps))
但这当然比显式转换要昂贵得多,顺便说一句,在像您描述的简单情况下,显式转换很容易实现自动化:
from pyspark.sql.functions import col
(spark
.createDataFrame(values, ("a", "date"))
.select([col(f.name).cast(f.dataType) for f in schema]))
关于python - 在模式中指定 DateType() 时从 RDD 创建 DataFrame,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/55038612/