在模式中指定 DateType() 时从 RDD 创建 DataFrame

Question

我正在从 RDD 创建一个 DataFrame，其中一个值是date. 我不知道如何DateType()在模式中指定。

让我来说明手头的问题——

我们可以将 加载date到 DataFrame 的一种方法是首先将其指定为字符串并date使用to_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)

有没有办法，我们可以DateType()在其中使用schema并避免显式转换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)

Answer 1

长话短说，与RDD外部对象一起使用的模式不打算以这种方式使用 - 声明的类型应该反映数据的实际状态，而不是所需的状态。

换句话说，允许：

schema = StructType([
  StructField('A', IntegerType(), True),
  StructField('date', DateType(), True)
])

与date字段对应的数据应使用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
)

最接近所需行为的是RDD[Row]使用 JSON 读取器转换数据 ( )，使用dicts

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]))