使用spark的 DataFrame 来操作mysql数据。
DataFrame是比RDD更高一个级别的抽象，可以应用SQL语句进行操作，详细参考：
https://spark.apache.org/docs/latest/sql-programming-guide.html
这里暂时使用spark-shell进行操作，
1.首先，必须要先下载一个mysql的jdbc的驱动
可以从这里下载
2.然后呢，就好办了。
#具体的启动spark-shell的方法（带上mysql的driver）
$~/spark-shell --driver-class-path /path-to-mysql-jar/mysql-connector-java-5.1.-bin.jar
#定义mysql的信息
val url="jdbc:mysql://10.181.176.226:3306/geo_info"
val prop = new java.util.Properties
prop.setProperty("user","geo")
prop.setProperty("password","xxxxxx”)
#指定读取条件,这里 Array("country_code='CN'") 是where过滤条件
val cnFlight = sqlContext.read.jdbc(url,"gps_location",Array("country_code='CN'"),prop)
#然后进行groupby 操作,获取数据集合
val emailList = cnFlight.groupBy("gps_city", "user_mail”)
#计算数目，并根据数目进行降序排序
val sorted = emailList.count().orderBy( desc("count") )
#显示前10条
sorted.show()
#存储到文件（这里会有很多分片文件。。。）
sorted.rdd.saveAsTextFile("/home/qingpingzhang/data/flight_top”)
#存储到mysql表里
sorted.write.jdbc(url,"table_name",prop)
3.具体文件编写代码，然后提交worker也类似，主要是DataFrame的 sqlContext声明会不一样。
val sc: SparkContext // An existing SparkContext.
val sqlContext = new org.apache.spark.sql.SQLContext(sc)
这里如果要用spark-submit，则会有坑，即便你是用sbt的assembly来打包的一个全的jar包：
参考：http://www.iteblog.com/archives/1300
[itelbog@iteblog ~]$ bin/spark-submit --master local[] --driver-class-path lib/mysql-connector-java-5.1..jar --class spark.SparkToJDBC ./spark-test_2.-1.0.jar