3

我正在尝试使用分桶技术对 Spark 作业进行一些性能优化。我正在阅读.parquet和.csv文件并进行一些转换。在我进行分桶并加入两个 DataFrame 之后。然后我正在将加入的 DF 写入镶木地板，但我有一个空文件~500B而不是500Mb.

  • Cloudera (cdh5.15.1)
  • 火花 2.3.0

  • 斑点

    val readParquet = spark.read.parquet(inputP)
readParquet
    .write
    .format("parquet")
    .bucketBy(23, "column")
    .sortBy("column")
    .mode(SaveMode.Overwrite)
    .saveAsTable("bucketedTable1")

val firstTableDF = spark.table("bucketedTable1")

val readCSV = spark.read.csv(inputCSV)
readCSV
    .filter(..)
    .ordrerBy(someColumn)

    .write
    .format("parquet")
    .bucketBy(23, "column")
    .sortBy("column")
    .mode(SaveMode.Overwrite)
    .saveAsTable("bucketedTable2")

val secondTableDF = spark.table("bucketedTable2")

val resultDF = secondTableDF
    .join(firstTableDF, Seq("column"), "fullouter")
    .
    .
resultDF
    .coalesce(1)
    .write
    .mode(SaveMode.Overwrite)
    .parquet(output)

ssh当我在命令行中使用我有正确的结果启动 Spark 作业时~500Mb，我可以使用 Hive 看到的镶木地板文件。如果我使用 oozie 工作流运行相同的作业，我有一个空文件 ( ~500 Bytes)。当我这样做.show()时，我resultDF可以看到数据，但我有空的镶木地板文件。

+-----------+---------------+----------+
|       col1|          col2 |      col3|
+-----------+---------------+----------+
|33601234567|208012345678910|       LOL|
|33601234567|208012345678910|       LOL|
|33601234567|208012345678910|       LOL|

当我不将数据保存为表格时，写入镶木地板没有问题。它仅发生在从表创建的 DF 中。

有什么建议么 ？

提前感谢您的任何想法！

4

1 回答 1

1

我为我的情况想通了，我刚刚添加了一个选项.option("path", "/sources/tmp_files_path")。现在我可以使用分桶，并且我的输出文件中有一个数据。

readParquet
  .write
  .option("path", "/sources/tmp_files_path")
  .mode(SaveMode.Overwrite)
  .bucketBy(23, "column")
  .sortBy("column")
  .saveAsTable("bucketedTable1")
于 2019-06-25T13:43:30.993 回答