随着分区的增长,火花拼花地板写得越慢

Question

我有一个火花流应用程序,从流中写入镶木地板数据.

sqlContext.sql(
      """
        |select
        |to_date(from_utc_timestamp(from_unixtime(at), 'US/Pacific')) as event_date,
        |hour(from_utc_timestamp(from_unixtime(at), 'US/Pacific')) as event_hour,
        |*
        |from events
        | where at >= 1473667200
      """.stripMargin).coalesce(1).write.mode(SaveMode.Append).partitionBy("event_date", "event_hour","verb").parquet(Config.eventsS3Path)

这段代码每小时运行一段时间,但随着时间的推移,对镶木地板的写作速度已经放慢.当我们开始时花了15分钟来写数据,现在需要40分钟.对于该路径中存在的数据而言,这需要时间.我尝试将相同的应用程序运行到新位置并且运行速度很快.

我已禁用schemaMerge和摘要元数据:

sparkConf.set("spark.sql.hive.convertMetastoreParquet.mergeSchema","false")
sparkConf.set("parquet.enable.summary-metadata","false")

使用spark 2.0

批处理执行:空目录 目录有350个文件夹

Answer 1

我遇到过这个问题。追加模式可能是罪魁祸首，因为随着镶木地板文件大小的增长，找到追加位置需要越来越多的时间。

我发现解决此问题的一种解决方法是定期更改输出路径。合并和重新排序所有输出数据帧中的数据通常不是问题。

def appendix: String = ((time.milliseconds - timeOrigin) / (3600 * 1000)).toString

df.write.mode(SaveMode.Append).format("parquet").save(s"${outputPath}-H$appendix")