有了上面四篇文章，再加上一些异常处理、优化，开发基本就没什么问题了。下面我们开始：

一：SparkSql异常处理

读JSON文件异常处理

　　　　val sparkSession= SparkSession.builder().master("local").getOrCreate()

　　　　var df2 = sparkSession.emptyDataFrame
　　　　try {
　　　　　　df2 = sparkSession.read.json("/JAVA/data/")
　　　　} catch {
　　　　　　case e: Exception => {
　　　　　　　　println("error info")
　　　　　　}
　　　　}
　　　　df2.show(100)

读CSV文件异常处理

　　　　val sparkSession= SparkSession.builder().master("local").getOrCreate()

　　　　var df2 = sparkSession.emptyDataFrame
　　　　try {
　　　　　　df2 = sparkSession.read.option("sep", "|").csv("/JAVA/data/")
　　　　　　　　.toDF("name","sex")
　　　　} catch {
　　　　　　case e: Exception => {
　　　　　　　　println("error info")
　　　　　　}
　　　　}
　　　　df2.show(100)

读TEXT文件异常处理。

　　　　个人理解CSV和TEXT一样，直接csv即可。还有一个原因是TEXT需要手动的去切分字符串作为一个列，使用起来太不方便了。还不如直接使用CSV

写文件异常

　　　　val sparkSession= SparkSession.builder().master("local").getOrCreate()
　　　　var df = sparkSession.emptyDataFrame
　　　　df = sparkSession.read.option("sep", "|").csv("/JAVA/data")
.　　　　　　toDF("name","sex")
　　　　df.write.mode(SaveMode.Overwrite).option("sep", "|").csv("/JAVA/data1")

 

　　　　SaveMode.Overwrite：覆盖式写文件，没有文件夹会创建文件夹

　　　　SaveMode.Append：添加式写文件，没有文件夹会报错，建议使用SaveMode.Overwrite

数据异常填充

　　　　进行真正开发的时候，经常join导致有一些空值（NULL），有时候产品需要将空值转换为一些特殊处理值：

　　　　val sparkSession= SparkSession.builder().master("local").getOrCreate()
　　　　val javasc = new JavaSparkContext(sparkSession.sparkContext)

　　　　val nameRDD = javasc.parallelize(Arrays.asList(
　　　　　　"{'name':'','age':''}",
　　　　　　"{'name':'sunliu','age':'19','vip':'true'}"));
　　　　val namedf = sparkSession.read.json(nameRDD)

　　　　namedf.na.fill(Map("name"->"zhangsan","age"->"18","vip"->"false")).show(100)//第一个数据不是空值，是空字符串

age	name	vip
		false
19	wangwu	true
19	wangwu	true

二：SparkSql优化

缓存

                     Spark中当一个Rdd多次使用的时候就需要进行缓存。缓存将大大的提高代码运行效率。

　　　　　　val sparkSession= SparkSession.builder().master("local").getOrCreate()
　　　　　　val javasc = new JavaSparkContext(sparkSession.sparkContext)

　　　　　　val nameRDD = javasc.parallelize(Arrays.asList(
　　　　　　　　"{'name':'','age':''}",
　　　　　　　　"{'name':'sunliu','age':'19','vip':'true'}"));
　　　　　　val namedf = sparkSession.read.json(nameRDD)
　　　　　　namedf.persist(StorageLevel.MEMORY_AND_DISK_SER)     

　　　　　　个人建议使用MEMORY_AND_DISK_SER，因为内存还是比较珍贵的，磁盘虽然慢但是大。

　　　　　　尽量不要使用MEMORY_AND_DISK_SER_2，这种后面有一个_2的，因为这是备份两个，一般情况下是不需要备份两个的。备份多了浪费内存。

Join策略

　　　　Spark有三种join的策略：broadcast join、Shuffle Hash Join、BroadcastHashJoin

　　　　broadcastHash join（大表和极小表）：

　　　　　　当大表join小表的时候：将小表进行广播到各个节点。

　　　　　　优点：不用进行数据shuffle，每个节点进行自己节点上数据的计算

　　　　　　缺点：将一个表的数据全部加载到主节点，对主节点的压力较大。

　　　　　　参数：广播的默认大小是10M可以适当将大小调整。 sparkSession.sql("set spark.sql.autoBroadcastJoinThreshold=134217728")

　　　　Shuffle Hash Join（大表和小表）

　　　　　　两个表进行重新分区之后，进行两个分区的数据遍历。

　　　　　　优点：分区之后数据更小了，就全部加载到内存遍历就行了

　　　　　　缺点：相对于broadcastHash join来说还是有一次shuffle

　　　　SortMergeJoin（大表和小表）

　　　　　　两个表进行重新分区之后，进行两个分区的数据遍历，个人感觉分区前和Shuffle Hash Join没什么区别。

　　　　　　缺点：分区之后数据还不能全部加载到内存，需要进行排序。将相同key的加载到内存。

执行计划

　　　　val sparkSession= SparkSession.builder().master("local").getOrCreate()
　　　　val javasc = new JavaSparkContext(sparkSession.sparkContext)

　　　　val nameRDD = javasc.parallelize(Arrays.asList("{'name':'wangwu','age':'18','vip':'t'}"));
　　　　val namedf = sparkSession.read.json(nameRDD)

　　　　namedf.explain()//显示执行计划

上线提交命令示例

　　　　spark-submit
　　　　--class class
　　　　--master yarn                                      
　　　　--executor-memory 6g                      //
　　　　--driver-memory 4g                           //
　　　　--num-executors 4                            //
　　　　--executor-cores 6                            //2~4个较为合适
　　　　--deploy-mode cluster                      //必须配置，默认是单个节点模式
　　　　--conf spark.driver.maxResultSize=6g
　　　　Jar.jar

Apache中文文档

                http://spark.apachecn.org/#/docs/7?id=spark-sql-dataframes-and-datasets-guide