PySpark 1.5 Cách cắt ngắn Dấu thời gian thành Phút gần nhất từ ​​giây

Question

Tôi đang sử dụng PySpark. Tôi có một cột ('dt') trong một khung dữ liệu ('canon_evt') rằng đây là một dấu thời gian. Tôi đang cố gắng để loại bỏ giây từ một giá trị DateTime. Nó ban đầu được đọc từ gỗ như một String. Sau đó, tôi cố chuyển đổi nó thành Dấu thời gian qua

canon_evt = canon_evt.withColumn('dt',to_date(canon_evt.dt)) 
canon_evt= canon_evt.withColumn('dt',canon_evt.dt.astype('Timestamp')) 

Sau đó, tôi muốn xóa giây. Tôi đã thử 'trunc', 'date_format' hoặc thậm chí cố ghép các mảnh lại với nhau như dưới đây. Tôi nghĩ rằng nó đòi hỏi một số loại bản đồ và sự kết hợp lambda, nhưng tôi không chắc chắn liệu Timestamp là một định dạng thích hợp, và cho dù nó có thể để thoát khỏi giây.

canon_evt = canon_evt.withColumn('dyt',year('dt') + '-' + month('dt') + 
    '-' + dayofmonth('dt') + ' ' + hour('dt') + ':' + minute('dt')) 

[Row(dt=datetime.datetime(2015, 9, 16, 0, 0),dyt=None)] 

Answer 1

Chuyển đổi sang timestamps Unix và arithmetics cơ bản nên để lừa:

from pyspark.sql import Row 
from pyspark.sql.functions import col, unix_timestamp, round 

df = sc.parallelize([ 
    Row(dt='1970-01-01 00:00:00'), 
    Row(dt='2015-09-16 05:39:46'), 
    Row(dt='2015-09-16 05:40:46'), 
    Row(dt='2016-03-05 02:00:10'), 
]).toDF() 


## unix_timestamp converts string to Unix timestamp (bigint/long) 
## in seconds. Divide by 60, round, multiply by 60 and cast 
## should work just fine. 
## 
dt_truncated = ((round(unix_timestamp(col("dt"))/60) * 60) 
    .cast("timestamp")) 

df.withColumn("dt_truncated", dt_truncated).show(10, False) 
## +-------------------+---------------------+ 
## |dt     |dt_truncated   | 
## +-------------------+---------------------+ 
## |1970-01-01 00:00:00|1970-01-01 00:00:00.0| 
## |2015-09-16 05:39:46|2015-09-16 05:40:00.0| 
## |2015-09-16 05:40:46|2015-09-16 05:41:00.0| 
## |2016-03-05 02:00:10|2016-03-05 02:00:00.0| 
## +-------------------+---------------------+ 

Answer 2

Tôi nghĩ zero323 có câu trả lời tốt nhất. Đó là loại gây phiền nhiễu mà Spark không hỗ trợ này nguyên bản, cho dễ dàng như thế nào là để thực hiện. Đối với hậu thế, dưới đây là một chức năng mà tôi sử dụng:

def trunc(date, format): 
    """Wraps spark's trunc fuction to support day, minute, and hour""" 
    import re 
    import pyspark.sql.functions as func 

    # Ghetto hack to get the column name from Column object or string: 
    try: 
     colname = re.match(r"Column<.?'(.*)'>", str(date)).groups()[0] 
    except AttributeError: 
     colname = date 

    alias = "trunc(%s, %s)" % (colname, format) 

    if format in ('year', 'YYYY', 'yy', 'month', 'mon', 'mm'): 
     return func.trunc(date, format).alias(alias) 
    elif format in ('day', 'DD'): 
     return func.date_sub(date, 0).alias(alias) 
    elif format in ('min',): 
     return ((func.round(func.unix_timestamp(date)/60) * 60).cast("timestamp")).alias(alias) 
    elif format in ('hour',): 
     return ((func.round(func.unix_timestamp(date)/3600) * 3600).cast("timestamp")).alias(alias)