关于spark写入文件至文件系统并制定文件名之自定义outputFormat

package cn.com.xxx.auditimport org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.hadoop.mapred.{FileOutputFormat, JobConf}
import org.apache.hadoop.mapred.lib.MultipleTextOutputFormatclass CustomOutputFormat extends MultipleTextOutputFormat[Any, Any] {

//重写generateFileNameForKeyValue方法，该方法是负责自定义生成文件的文件名
override def generateFileNameForKeyValue(key: Any, value: Any, name: String): String = {
//这里的key和value指的就是要写入文件的rdd对，再此，我定义文件名以key.txt来命名，当然也可以根据其他的需求来进行生成文件名
val fileName = key.asInstanceOf[String] + ".txt"
fileName
}
/**
*因为saveAsHadoopFile是以key,value的形式保存文件，写入文件之后的内容也是，按照key value的形式写入，k,v之间用空格隔开，这里我只需要写入value的值，不需要将key的值写入到文件中个，所以我需要重写
*该方法，让输入到文件中的key为空即可，当然也可以进行领过的变通，也可以重写generateActuralValue(key:Any,value:Any),根据自己的需求来实现
*/
override def generateActualKey(key: Any, value: Any): String = {
null
}
//对生成的value进行转换为字符串，当然源码中默认也是直接返回value值，如果对value没有特殊处理的话，不需要重写该方法
override def generateAcutalValue(key: Any, value: Any): String = {
return value.asInstance[String]
}
/**
* 该方法使用来检查我们输出的文件目录是否存在，源码中，是这样判断的，如果写入的父目录已经存在的话，则抛出异常
* 在这里我们冲写这个方法，修改文件目录的判断方式，如果传入的文件写入目录已存在的话，直接将其设置为输出目录即可，
* 不会抛出异常
*/
override def checkOutputSpecs(ignored: FileSystem, job: JobConf): Unit = {
var outDir: Path = FileOutputFormat.getOutputPath(job)
if (outDir != null) {
//注意下面的这两句，如果说你要是写入文件的路径是hdfs的话，下面的两句不要写，或是注释掉，它俩的作用是标准化文件输出目录，根据我的理解是，他们是标准化本地路径，写入本地的话，可以加上，本地路径记得要用file:///开头，比如file:///E:/a.txt
//val fs: FileSystem = ignored
//outDir = fs.makeQualified(outDir)
FileOutputFormat.setOutputPath(job, outDir)
}
}
}

package scala.spark._sqlimport java.util.Propertiesimport mysqlUtils.OperatorMySql
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.hadoop.mapred.lib.MultipleTextOutputFormat
import org.apache.hadoop.mapred.{FileOutputFormat, JobConf}
import org.apache.spark.sql.{DataFrame, SQLContext}
import org.apache.spark.{SparkConf, SparkContext, TaskContext}
object DataFrameToMySql {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setAppName(this.getClass.getName).setMaster("local[2]")
val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc)
//配置输出文件不生成success文件
sc.hadoopConfiguration.set("mapreduce.fileoutputcommitter.marksuccessfuljobs", "false")
//配置一些参数
//如果设置为true，sparkSql将会根据数据统计信息，自动为每一列选择单独的压缩编码方式
sqlContext.setConf("spark.sql.inMemoryColumnarStorage.compressed", "true")
//控制列式缓存批量的大小。增大批量大小可以提高内存的利用率和压缩率，但同时也会带来OOM的风险
sqlContext.setConf("spark.sql.inMemoryColumnarStorage.batchSize", "1000")
sqlContext.setConf("spark.sql.autoBroadcastJoinThreshold", "10485760")
//设为true，则启用优化的Tungsten物理执行后端。Tungsten会显示的管理内存，并动态生成表达式求值得字节码
sqlContext.setConf("spark.sql.tungsten.enabled", "true")
//配置shuffle是的使用的分区数
sqlContext.setConf("spark.sql.shuffle.partitions", "200")sc.setLogLevel("WARN")
val pro = new Properties()
pro.put("user", "root")
pro.put("password", "123456")
pro.put("driver", "com.mysql.jdbc.Driver")
val url = "jdbc:mysql://localhost:3306/test?serverTimezone=UTC"
val rdf = sqlContext.read /*.jdbc(url,"person1",pro)*/
.format("jdbc")
.options(Map(
"url" -> url,
"dbtable" -> "person",
"driver" -> "com.mysql.jdbc.Driver",
"user" -> "root",
"password" -> "123456",
"fetchSize" -> "10",
"partitionColumn" -> "age",
"lowerBound" -> "0",
"upperBound" -> "1000",
"numPartitions" -> "2"
)).load()
//将读取的文件尽心个计算，并且以pairRDD的形式写入文件中，这里在写入文件的时候，会将key当做文件名来进行写入，也就是说相同的key对应的value都会写入到相同的文件中
val x = rdf.groupBy(substring(col("score"), 0, 5) as ("score")).agg(max("age") as ("max"), avg("age") as ("avg"))
.rdd.map(x => ("person", x(0) + "," + x(1) + "," + x(2)))
//这里partitionBy，只是来增加文件文件写入的并行度，可以根据需求进行设置，影响的是文件写入的性能，我个人是这么理解的，如果有不对的还请指正
.partitionBy(new HashPartitioner(10))
//这里写入的时候，要指定我们自定义的PairRDDMultipleTextOutputFormat类
.saveAsHadoopFile("file:///E:/dataFile/res", classOf[String], classOf[String], classOf[PairRDDMultipleTextOutputFormat])
sc.stop()
}