package cn.ac.iie.logimport org.apache.spark.sql.{SaveMode, SparkSession}/*** 使用Spark完成我們的數(shù)據(jù)清洗操作*/ object SparkStatCleanJob {def main(args: Array[String]): Unit = {val spark = SparkSession.builder().appName("SparkStatCleanJob").master("local[2]").getOrCreate()val acessRDD = spark.sparkContext.textFile("file:///E:/test/output/part-00000")// acessRDD.take(10).foreach(println)// RDD => DFval accessDF = spark.createDataFrame(acessRDD.map(x => AccessConvertUtil.parseLog(x)), AccessConvertUtil.struct)// accessDF.printSchema()// accessDF.show(false)accessDF.coalesce(1).write.format("parquet").mode(SaveMode.Overwrite).partitionBy("day").save("file:///E:/test/clean")spark.stop()}}

有時我們要將DataFrame保存到文件系統(tǒng)中，并且按照day字段進(jìn)行partitionBy，往往使用accessDF.write.format("parquet").mode(SaveMode.Overwrite).partitionBy("day").save("file:///E:/test/clean")這種方式，
這句話：將DataFrame數(shù)據(jù)以parquet文件格式寫入file:///E:/test/clean文件系統(tǒng)中，并且按照day進(jìn)行partition，同時如果文件夾存在則進(jìn)行覆蓋。
但是有時候，輸出的結(jié)果中day文件夾下的小文件太多了，非常影響性能。因此使用coalesce(1)來將小文件文件輸出到一個文件中。之前在一個day下會有多個文件，使用accessDF.coalesce(1).write.format("parquet").mode(SaveMode.Overwrite).partitionBy("day").save("file:///E:/test/clean")方式之后，一個day下就僅有一個文件了。

總結(jié)

以上是生活随笔為你收集整理的SparkSQL 控制文件输出的大小的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。