今天簡單講解一下應用spark1.5.2相關讀取mysql資料到DataFrame的介面以及將DF資料存放到mysql中介面實現例項

同樣我們的程式設計開發環境是不需要安裝spark的，但是需要一臺安裝了mysql的伺服器，我這裡直接在本機安裝了一個mysql，還有就是scala的程式設計環境。

注意本次使用的spark版本是1.5.2，相關引用的包請參考下圖：

先看程式碼吧

package JDBC_MySql

import java.util.Properties

import org.apache.spark.sql.SaveMode
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by zhoubh on 2016/7/20.
  */
object mysqlDB {

  case class zbh_test(day_id:String, prvnce_id:String,pv_cnts:Int)

  def main(args: Array[String]) {


    val conf = new SparkConf().setAppName("mysql").setMaster("local[4]")
    val sc = new SparkContext(conf)
    //sc.addJar("D:\\workspace\\sparkApp\\lib\\mysql-connector-java-5.0.8-bin.jar")
    val sqlContext = new org.apache.spark.sql.SQLContext(sc)



     //定義mysql資訊
    val jdbcDF = sqlContext.read.format("jdbc").options(
      Map("url"->"jdbc:mysql://localhost:3306/db_ldjs",
    "dbtable"->"(select imei,region,city,company,name from tb_user_imei) as some_alias",
    "driver"->"com.mysql.jdbc.Driver",
    "user"-> "root",
    //"partitionColumn"->"day_id",
    "lowerBound"->"0",
    "upperBound"-> "1000",
    //"numPartitions"->"2",
    "fetchSize"->"100",
    "password"->"123456")).load()


    jdbcDF.collect().take(20).foreach(println) //終端列印DF中的資料。
    //jdbcDF.rdd.saveAsTextFile("C:/Users/zhoubh/Downloads/abi_sum")
    val url="jdbc:mysql://localhost:3306/db_ldjs"
    val prop=new Properties()
    prop.setProperty("user","root")
    prop.setProperty("password","123456")
    jdbcDF.write.mode(SaveMode.Overwrite).jdbc(url,"zfs_test",prop) //寫入資料庫db_ldjs的表 zfs_test 中
    //jdbcDF.write.mode(SaveMode.Append).jdbc(url,"zbh_test",prop)  //你會發現SaveMode改成Append依然無濟於事，表依然會被重建，為了解決這個問題，後期會另開部落格講解

     //org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils.saveTable(jdbcDF,url,"zbh_test",prop)
////    #然後進行groupby 操作,獲取資料集合
//    val abi_sum_area = abi_sum.groupBy("date_time", "area_name")
//
////    #計算數目，並根據數目進行降序排序
//    val sorted = abi_sum_area.count().orderBy("count")
//
////    #顯示前10條
//    sorted.show(10)
//
////    #儲存到檔案（這裡會有很多分片檔案。。。）
//    sorted.rdd.saveAsTextFile("C:/Users/zhoubh/Downloads/sparktest/flight_top")
//
//
////    #儲存到mysql表裡
//    //sorted.write.jdbc(url,"table_name",prop)

  }
}
 

下面來看看執行結果啥樣：

資料庫結果如下：

通過這段程式碼可以實現從mysql關係型資料庫中直接讀取資料轉化成DataFrame參與到sparksql的分析當中這個意義是非常重大的，因為我們日常應用sparksql進行資料分析時經常會用到一些配置表，而這些配置定義表都是存在關係型資料庫中，所以以後不用擔心了。

另外這裡還實現了DataFrame結果回寫到mysql資料庫中，雖然官方的spark原始碼的寫入有些奇葩，設定的寫死模式overwriter，也就是說你確定寫入的表，他會重新建立，然後匯入資料，這個用起來很不爽，後面部落格將講解如何改寫原始碼，我要怎麼寫入就怎麼寫入。（這個意義也很重大，以後分析的結果就可以直接放mysql中，直接對外提供報表，哇 贊

最後感覺華哥的程式碼和講解（一個個默默耕耘大資料多年的人）