我正在使用Spark 2.1.0。 我无法在Pyspark中创建时间戳列,我正在使用以下代码片段。请帮忙。
我正在使用Spark 2.1.0,但是以下代码片段无法创建Pyspark中的时间戳列。请协助解决。
df=df.withColumn('Age',lit(datetime.now()))
我遇到了问题:
断言错误:col 应该是列(Column)
请帮忙解决。
我不确定2.1.0,但至少在2.2.1中,您可以简单地执行以下操作:
from pyspark.sql import functions as F
df.withColumn('Age', F.current_timestamp())
希望能对您有所帮助!
>>> dict = [{'name': 'Alice', 'age': 1},{'name': 'Again', 'age': 2}]
>>> df = spark.createDataFrame(dict)
>>> import time
>>> import datetime
>>> timestamp = datetime.datetime.fromtimestamp(time.time()).strftime('%Y-%m-%d %H:%M:%S')
>>> type(timestamp)
<class 'str'>
>>> from pyspark.sql.functions import lit,unix_timestamp
>>> timestamp
'2017-08-02 16:16:14'
>>> new_df = df.withColumn('time',unix_timestamp(lit(timestamp),'yyyy-MM-dd HH:mm:ss').cast("timestamp"))
>>> new_df.show(truncate = False)
+---+-----+---------------------+
|age|name |time |
+---+-----+---------------------+
|1 |Alice|2017-08-02 16:16:14.0|
|2 |Again|2017-08-02 16:16:14.0|
+---+-----+---------------------+
>>> new_df.printSchema()
root
|-- age: long (nullable = true)
|-- name: string (nullable = true)
|-- time: timestamp (nullable = true)
在 balalaika 的基础上,如果有人像我一样只想添加日期而不是时间,那么可以按照下面的代码操作:
from pyspark.sql import functions as F
df.withColumn('Age', F.current_date())