Spark SQLクエリーをプログラミングで実行

sparkを初期化するには、2つのパラメータを渡すだけです.

val conf = new SparkConf().setMaster("local").setAppName("My App")
val sc = new SparkContext(conf)

クラスタURL:Sparkがクラスタにどのように接続されているかを示します.このいくつかの例ではlocalを使用しています.この特殊な値は、クラスタに接続することなく、Sparkをスタンドアロン単一スレッド上で実行することができます.

アプリケーション名:例ではMy Appを使用しています.クラスタに接続すると、この値はクラスタマネージャのユーザーインタフェースでアプリケーションを見つけるのに役立ちます.

 

package cn.itcast.spark.sql

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

object InferringSchema {
  def main(args: Array[String]) {

    //  SparkConf()   App  
    val conf = new SparkConf().setAppName("SQL-1")
    //SQLContext   SparkContext
    val sc = new SparkContext(conf)
    //  SQLContext
    val sqlContext = new SQLContext(sc)

    //        RDD
    val lineRDD = sc.textFile(args(0)).map(_.split(" "))

    //  case class
    // RDD case class  
    val personRDD = lineRDD.map(x => Person(x(0).toInt, x(1), x(2).toInt))
    //      ，        RDD   DataFrame
    // RDD   DataFrame
    import sqlContext.implicits._
    val personDF = personRDD.toDF
    //   
    personDF.registerTempTable("t_person")
    //  SQL
    val df = sqlContext.sql("select * from t_person order by age desc limit 2")
    //    JSON          
    df.write.json(args(1))
    //  Spark Context
    sc.stop()
  }
}
//case class       
case class Person(id: Int, name: String, age: Int)

データソース:

1. mysql     
val jdbcDF = sqlContext.read.format("jdbc").options(Map("url" -> "jdbc:mysql://192.168.10.1:3306/bigdata", "driver" ->      "com.mysql.jdbc.Driver", "dbtable" -> "person", "user" -> "root", "password" -> "123456")).load()
2.     
jdbcDF.show()

mysqlデータベースにデータを書き込みます.

package cn.itcast.spark.sql

import java.util.Properties
import org.apache.spark.sql.{SQLContext, Row}
import org.apache.spark.sql.types.{StringType, IntegerType, StructField, StructType}
import org.apache.spark.{SparkConf, SparkContext}

object JdbcRDD {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("MySQL-Demo")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    //       RDD
    val personRDD = sc.parallelize(Array("1 tom 5", "2 jerry 3", "3 kitty 6")).map(_.split(" "))
    //  StructType         schema
    val schema = StructType(
      List(
        StructField("id", IntegerType, true),
        StructField("name", StringType, true),
        StructField("age", IntegerType, true)
      )
    )
    // RDD   rowRDD
    val rowRDD = personRDD.map(p => Row(p(0).toInt, p(1).trim, p(2).toInt))
    // schema     rowRDD 
    val personDataFrame = sqlContext.createDataFrame(rowRDD, schema)
    //  Properties         
    val prop = new Properties()
    prop.put("user", "root")
    prop.put("password", "123456")
    //         
    personDataFrame.write.mode("append").jdbc("jdbc:mysql://192.168.10.1:3306/bigdata", "bigdata.person", prop)
    //  SparkContext
    sc.stop()
  }
}