小白學習Spark系列四：RDD踩坑總結（scala+spark2.1 sql常用方法）

　　初次嘗試用 Spark+scala 完成項目的重構，由於兩者之前都沒接觸過，所以邊學邊用的過程大多艱難。首先面臨的是如何快速上手，然后是代碼調優、性能調優。本章主要記錄自己在項目中遇到的問題以及解決方式，下篇會嘗試調優方法。末尾會分享自己的學習資料，也供大多菜鳥第一次使用作為參考。由於自己項目中大量使用spark sql，所以下面的經驗大多是和spark sql有關。同樣下面也列出作為菜鳥在學習過程中的困惑以及踩的坑，還請大牛勿笑 ~_~ 如果有更好的方式解決，歡迎留言，一起學習。

1、常用場景

（1）場景一：rdd讀取指定行分隔符的數據，不以每行為單位

例1：配置文件中有n個sql語句，每個sql以分號----分隔。你需要讀取sql，分別從hdfs中拉取數據。可能會采取：

//conf_sql_map_file 是sql配置文件
val sql_rdd = sc.textFile(conf_sql_map_file)
var sqls = sql_rdd.collect().mkString(" ").split("----")

分析：由於rdd以每行為單位，自動去掉結尾的 換行符。但sql配置文件需要以指定分隔符分隔，而不是每行。所以使用 mkString(" ") 將讀取的每行數據以空格分隔，整合為一個長字符串，最后以分隔符分隔。

 

但如果 sql 語句中有使用 with 之類的關鍵詞時，上面那種方式讀取配置文件會因為格式問題會出錯，with語句需要和 select 語句空行分隔，為保險起見，以 “\n” 分隔，還原配置文件的原始格式。

var sqls = sql_rdd.collect().mkString("\n").split("----")

 

（2）場景二：讀取文件，以 key-value 形式存儲。

例2：文件file1內容如下

key1,value1

key2,value2

var file_rdd = sc.textFile(file1).map(e=> (e.split(',')(0),e.split(',')(1))).collectAsMap

或者 不從文件讀取，直接使用List類型數據演示

scala> var line_rdd = sc.parallelize(List[String]("k,v","key,value")).map(e=>(e.split(',')(0),e.split(',')(1))).collectAsMap
line_rdd: scala.collection.Map[String,String] = Map(k -> v, key -> value)

 分析：collectAsMap 是行動操作的一種，可以將數據類型轉換為Map類型，而collect是直接轉為Array類型。

 

（3）場景三：從hive表中讀取數據放到array數組中，其中每條數據轉換為List類型。

scala> import org.apache.spark.{SparkConf, SparkContext}
scala> import org.apache.spark.sql.SparkSession

scala> val conf = new SparkConf().setAppName("graph_spark@zky")
//設置本程序名稱
scala> val hiveCtx: SparkSession = SparkSession.builder.config(conf).enableHiveSupport().getOrCreate()
//使用rdd函數轉換格式
scala> var sql_file_result = hiveCtx.sql("select * from city limit 10").rdd

scala> sql_file_result
res10: org.apache.spark.rdd.RDD[org.apache.spark.sql.Row] = MapPartitionsRDD[1187] at rdd at <console>:29

scala> sql_file_result.first
res11: org.apache.spark.sql.Row = [110000,北京市,110000,1,-911,2015-10-10 12:09:47,-911,2018-01-09 18:27:28,20181001000000]

分析：由於spark2.0版本丟棄了SQLContext(HiveContext)，取而代之的是SparkSession。hdfs拉取的數據格式為 org.apache.spark.sql.Row，需要調用mkString("\t") 對其轉換為String類型的rdd ，然后再轉換為其他類型。

但當你的數據以制表符分隔，就像下面代碼里一樣，末尾字段值如果存在字符串""空時，建議在首尾加上 [ ] 標識符，因為制表符和末尾的空值都會被rdd 自動過濾掉。另外，不建議分隔符使用制表符分隔，在選用分隔符時確保數據中不會出現你指定的分隔符。

scala> var lines = sql_file_result.map(line => "["+line.mkString("\t")+"]")
lines: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[1189] at map at <console>:33

scala> lines.collect
res14: Array[String] = Array([110000	北京市	110000	1	-911	2015-10-10 12:09:47	-911	2018-01-09 18:27:28	20180123000000], [120000	天津市	120000	1	-911	2015-10-10 12:09:47	-911	2018-01-09 18:27:28	20180123000000],。。。

 

解析帶[ ]的字符串轉成list格式，split()函數中的-1是為確保空值不被過濾。

scala> var items = lines.map(line => line.substring(1,line.length-1).split("\t",-1).toList)
items: org.apache.spark.rdd.RDD[List[String]] = MapPartitionsRDD[1190] at map at <console>:35

scala> items.collect
res15: Array[List[String]] = Array(List(370101, 濟南市, 370000, 1, -911, 1000-01-01 00:00:00, -911, 1000-01-01 00:00:00, 20180916000000), List(110000, 北京市, 110000, 1, -911, 2015-10-10 12:09:47, -911, 2018-01-09 18:27:28, 20180916000000),

　　

　　（4）場景四：從hive表中讀取的數據存儲為Map映射。

scala> var mid_data_rdd = hiveCtx.sql("select city_code,city_name from city limit 10").rdd
scala> mid_data_rdd.collect
res16: Array[org.apache.spark.sql.Row] = Array([110000,北京市], [120000,天津市], [130100,石家庄市], [130200,唐山市], [130300,秦皇島市], [130400,邯鄲市], [130500,邢台市], [130600,保定市], [130700,張家口市], [130800,承德市])

scala> var mid_data_map = mid_data_rdd.map(x => (x(0)->x(1).toString)).collectAsMap
mid_data_map: scala.collection.Map[Any,String] = Map(110000 -> 北京市, 130100 -> 石家庄市, 130300 -> 秦皇島市, 120000 -> 天津市, 130500 -> 邢台市, 130700 -> 張家口市, 130200 -> 唐山市, 130400 -> 邯鄲市, 130600 -> 保定市, 130800 -> 承德市)

scala> var mid_data_map = mid_data_rdd.map(x => (x(0).toString->x(1).toString)).collectAsMap
mid_data_map: scala.collection.Map[String,String] = Map(130300 -> 秦皇島市, 130600 -> 保定市, 130500 -> 邢台市, 130800 -> 承德市, 130200 -> 唐山市, 110000 -> 北京市, 130400 -> 邯鄲市, 130700 -> 張家口市, 130100 -> 石家庄市, 120000 -> 天津市)

//如果想轉換為array數組，試一下collect~

scala> var mid_data_map = mid_data_rdd.map(x => (x(0).toString->x(1).toString)).collect
mid_data_map: Array[(String, String)] = Array((110000,北京市), (120000,天津市), (130100,石家庄市), (130200,唐山市), (130300,秦皇島市), (130400,邯鄲市), (130500,邢台市), (130600,保定市), (130700,張家口市), (130800,承德市))

分析：可以關注下 toString函數~

（5）場景五：將數據通過寫入臨時表以存儲到hive表

scala> val people = sc.parallelize(List(("1","mary"),("2","rose"),("3","jack")))
people: org.apache.spark.rdd.RDD[(String, String)] = ParallelCollectionRDD[1] at parallelize at <console>:24
scala> case class Person(id:String,name:String)
defined class Person
scala> var people_trans = people.map(item => Person(item._1,item._2))
people_trans: org.apache.spark.rdd.RDD[Person] = MapPartitionsRDD[2] at map at <console>:28
scala> val people_frame = people_trans.toDF()
people_frame: org.apache.spark.sql.DataFrame = [id: string, name: string]

scala> people_frame.createOrReplaceTempView("person")
scala> import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.SparkSession

scala> import org.apache.spark.SparkConf
import org.apache.spark.SparkConf

scala> val conf = new SparkConf().setAppName("graph_spark@zhengkaiyu")
conf: org.apache.spark.SparkConf = org.apache.spark.SparkConf@534df4b

scala> val hiveCtx: SparkSession = SparkSession.builder.config(conf).enableHiveSupport().getOrCreate()
18/11/19 21:47:11 WARN SparkSession$Builder: Using an existing SparkSession; some configuration may not take effect.
hiveCtx: org.apache.spark.sql.SparkSession = org.apache.spark.sql.SparkSession@18de437d

scala> import hiveCtx.sql
import hiveCtx.sql

scala> import hiveCtx.implicits._
import hiveCtx.implicits._

scala> sql("select * from person").collect
res6: Array[org.apache.spark.sql.Row] = Array([1,mary], [2,rose], [3,jack])   

scala> sql("insert into 庫名.hive表名 select * from person")

　　分析：此例是基於case class來創建SchemaRDD，通過寫入臨時表，最后再插入到hive表中。除了這種方式還可以基於json格式來建臨時表，見下例。其中spark2.1建立臨時表時，將registerTempTable() 改為createOrReplaceTempView(),注意版本，要不會引起不必要的麻煩。

scala> import org.apache.spark.sql.SparkSession
scala> import org.apache.spark.SparkConf
scala> val conf = new SparkConf().setAppName("graph_spark@zhengkaiyu")
scala> val spark: SparkSession = SparkSession.builder.config(conf).enableHiveSupport().getOrCreate()
scala> val df = spark.read.json("examples/src/main/resources/people.json")
scala> df.show()
scala> df.createOrReplaceTempView("people")
scala> val sqlDF = spark.sql("SELECT * FROM people")
scala> sqlDF.show()

2、注意事項

（1）當啟動交互環境 spark-shell 時，會出現較為詭異的事情，剛定義好的變量會被之前的同名變量所覆蓋，猜想原因可能是內存不足導致。

（2）在scala代碼中，建議if-else語句格式規范書寫，否則會編譯不正確。

if(條件){

}

（3）啟動 spark-shell 時，注意指定的模式local、yarn。

 

3、常見錯誤的解決方法

不可序列化：org.apache.spark.SparkException: Task not serializable

解決方案1：繼承java可序列化類

object Process extends java.io.Serializable{
}

參考博客：https://stackoverflow.com/questions/22592811/task-not-serializable-java-io-notserializableexception-when-calling-function-ou

通過從臨時表中讀取數據寫入hive表時，會遇到錯誤：org.apache.spark.SparkException: Dynamic partition strict mode requires at least one static partition column. To turn this off set hive.exec.dynamic.partition.mode=nonstrict

        解決方案：執行下面語句后再執行insert語句。

//val spark: SparkSession = SparkSession.builder.config(conf).enableHiveSupport().getOrCreate()

spark.sql("SET hive.exec.dynamic.partition = true") 
spark.sql("SET hive.exec.dynamic.partition.mode = nonstrict ") 
spark.sql("SET hive.exec.max.dynamic.partitions.pernode = 400")

　　

 

4、學習資料

《Spark快速大數據分析》王道遠譯，推薦理由：快速上手，實例代碼有python、scala、java三種語言

《快學scala》

https://spark.apache.org/docs/2.1.0/sql-programming-guide.html#datasets-and-dataframes

https://tech.meituan.com/spark_tuning_pro.html

http://dblab.xmu.edu.cn/blog/spark-quick-start-guide/#more-577

spark.sql數據類型：http://spark.apache.org/docs/1.3.1/api/scala/index.html#org.apache.spark.sql.Row