过滤并保存数据框的前 X 行答案

【问题标题】：filter and save first X lines of a dataframe过滤并保存数据框的前 X 行
【发布时间】：2017-03-18 09:04:50
【问题描述】：

我正在使用 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 匹配一些预定义的值。

我的意图是将这些行保存到 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(....)

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

【问题讨论】：

标签： apache-spark pyspark spark-dataframe spark-csv

【解决方案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")

【讨论】：

谢谢，但这不是必需的，因为我有自己的encode_time UDF，如您所见（它返回日期+小时的字符串，因为这是我的要求）。关于filter, limit 我也设法用这些解决了它，你错过了我也在寻找的option("header", "true") :)

【解决方案2】：

使用filter()和limit()方法解决如下：

new_log_df.filter(col('timePeriod') == '20161206, Morning').limit(50).write.\
    format('csv').option("header", "true").save("..Path..")

【讨论】：