PySpark从TimeStampType列向DataFrame添加一列 | park从TimeStampType列向DataFrame添加一

本文介绍了PySpark从TimeStampType列向DataFrame添加一列的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！

问题描述

我有一个看起来像这样的DataFrame.我想在date_time字段的当天进行操作.

I have a DataFrame that look something like that. I want to operate on the day of the date_time field.

root
 |-- host: string (nullable = true)
 |-- user_id: string (nullable = true)
 |-- date_time: timestamp (nullable = true)

我尝试添加一列以提取日期.到目前为止，我的尝试都失败了.

I tried to add a column to extract the day. So far my attempts have failed.

df = df.withColumn("day", df.date_time.getField("day"))

org.apache.spark.sql.AnalysisException: GetField is not valid on fields of type TimestampType;

这也失败了

df = df.withColumn("day", df.select("date_time").map(lambda row: row.date_time.day))

AttributeError: 'PipelinedRDD' object has no attribute 'alias'

有什么想法可以做到吗?

Any idea how this can be done?

推荐答案

您可以使用简单的map:

df.rdd.map(lambda row:
    Row(row.__fields__ + ["day"])(row + (row.date_time.day, ))
)

另一种选择是注册一个函数并运行SQL查询:

Another option is to register a function and run SQL query:

sqlContext.registerFunction("day", lambda x: x.day)
sqlContext.registerDataFrameAsTable(df, "df")
sqlContext.sql("SELECT *, day(date_time) as day FROM df")

最后，您可以像这样定义udf:

Finally you can define udf like this:

from pyspark.sql.functions import udf
from pyspark.sql.types import IntegerType

day = udf(lambda date_time: date_time.day, IntegerType())
df.withColumn("day", day(df.date_time))

编辑:

实际上，如果您使用原始SQL，则已经定义了day函数(至少在Spark 1.4中)，因此您可以省略udf注册.它还提供了许多不同的日期处理功能，包括:

Actually if you use raw SQL day function is already defined (at least in Spark 1.4) so you can omit udf registration. It also provides a number of different date processing functions including:

类似 year ， month ， dayofmonth

日期算术工具，例如， datediff

date arithmetics tools like date_add, datediff

还可以使用简单的日期表达式，例如:

It is also possible to use simple date expressions like:

current_timestamp() - expr("INTERVAL 1 HOUR")

这意味着您可以构建相对复杂的查询，而无需将数据传递给Python.例如:

It mean you can build relatively complex queries without passing data to Python. For example:

df =  sc.parallelize([
    (1, "2016-01-06 00:04:21"),
    (2, "2016-05-01 12:20:00"),
    (3, "2016-08-06 00:04:21")
]).toDF(["id", "ts_"])

now = lit("2016-06-01 00:00:00").cast("timestamp")
five_months_ago = now - expr("INTERVAL 5 MONTHS")

(df
    # Cast string to timestamp
    # For Spark 1.5 use cast("double").cast("timestamp")
    .withColumn("ts", unix_timestamp("ts_").cast("timestamp"))
    # Find all events in the last five months
    .where(col("ts").between(five_months_ago, now))
    # Find first Sunday after the event
    .withColumn("next_sunday", next_day(col("ts"), "Sun"))
    # Compute difference in days
    .withColumn("diff", datediff(col("ts"), col("next_sunday"))))

这篇关于PySpark从TimeStampType列向DataFrame添加一列的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！