3

我正在使用 pySpark 读取和计算数据帧的统计信息。

数据框如下所示：

TRANSACTION_URL    START_TIME        END_TIME           SIZE    FLAG  COL6 COL7 ...
www.google.com     20170113093210    20170113093210     150      1    ...  ...
www.cnet.com       20170113114510    20170113093210     150      2    ...  ...

我正在向数据框中添加一个新timePeriod列，添加后，我想保存前 50K 记录并timePeriod匹配一些预定义的值。

我的意图是使用 dataframe header将这些行保存到 CSV中。

我知道这应该是和的组合，col但write.csv我不确定如何正确使用它们来实现我的意图。

我目前的代码是：

encodeUDF = udf(encode_time, StringType())
log_df = log_df.withColumn('timePeriod', encodeUDF(col('START_TIME')))

在添加列之后，我猜我应该使用类似的东西：

log_df.select(col('timePeriod') == 'Weekday').write.csv(....)

有人可以帮我在这里填空，以符合我的意图吗？

4

2 回答 2

1

unix_timestamp并且date_format在这里是有用的方法，因为START_TIME不是时间戳类型。

dfWithDayNum = log_df.withColumn("timePeriod", date_format(
  unix_timestamp(col("START_TIME"), "yyyyMMddHHmmss").cast(TimestampType), "u")
)

timePeriod将有星期几（1 = 星期一，...，7 = 星期日）

dfWithDayNum
  .filter(col("timePeriod") < 6) //to filter weekday
  .limit(50000) //X lines
  .write.format("csv")
  .option("header", "true")
  .csv("location/to/save/df")
于 2017-03-18T12:31:30.690 回答
0

通过以下方式使用filter()和方法解决：limit()

new_log_df.filter(col('timePeriod') == '20161206, Morning').limit(50).write.\
    format('csv').option("header", "true").save("..Path..")
于 2017-03-18T12:30:58.560 回答