spark sql with json and xml processing

-----------


   Spark Sql
---------------

  [ ]
     Spark sql is a library,
   to process spark data objects,
   using sql select statements.

 Spark sql follows mysql based sql syntaxes.
==============================================
 Spark sql provides,
   two types of contexts.
  i) sqlContext
 ii) HiveContext.

import org.apache.spark.sql.SqlContext

    val sqlCon = new SqlContext(sc)

using sqlContext ,
   we can process spark objects using select statements.

 Using HiveContext,
   we can integrate , Hive with Spark.
 Hive, is data warehouse environment in hadoop framework,
   So total is stored and managed at Hive tables.
  using HiveContext we can access entire hive enviroment (hive tables) from Spark.

  difference between, hql statement from Hive,
  and hql statement from Spark.
--> if hql is executed from Hive Environment,
  the statement to process, will be converted
 as mAPREDUCE job.
 ---> if same hive is integrated with spark,
  and hql is submitted from spark,
    it uses, DAG and Inmemory computing models.
  which is more faster than MapReduce.

 import org.apache.spark.sql.hive.HiveContext

 val hc = new HiveContext(sc)
-----------------------------
 Example of sqlContext.

 val sqc = new SqlContext(sc)

  file name --> file1
  sample --->  100,200,300
               300,400,400
                :
                :
 step1)
  create case class for the data.
 
   case class  Rec(a:Int, b:Int, c:Int)

 step2) create a function ,
   to convert raw line into case object.
  [function to provide schema ]

 def makeRec(line:String)={
        val w = line.split(",")
        val a = w(0).toInt
        val b = w(1).toInt
        val c = w(2).toInt
        val r = Rec(a, b,c)
         r
    }
--------
step3) load data.
   val data = sc.textFile("/user/cloudera/sparklab/file1")

   100,200,300
   2000,340,456
    :
    :


step4) transform each record into case Object

  val recs = data.map(x => makeRec(x))

step5) convert rdd into data frme.
 
  val df = recs.toDF

step6)  create table instance for the dataframe.

   df.registerTempTable("samp")

step7) apply select statement of sql on temp table.

val  r1 = sqc.sql("select a+b+c as tot from samp")

    r1
  ------
   tot
   ----
   600
   900

  r1.registerTempTable(samp1)

val r2 =    sqc.sql("select sum(tot) as gtot from samp1")


   once "select" statement is applied on
 temp table, returned object will be dataframe.


 to apply sql on processed results,
    again we need to register the dataframe
  as temp table.

 r1.registerAsTempTable("Samp2")
 
 val r2 = sqc.sql("select * from samp2
                      where tot>=200")

-----------------------------------

         sales
   --------------------
     :
   12/27/2016,10000,3,10
     :
     :

  -------------------------
 Steps involing in Spark Sql.[sqlContext]
----------------------------
   monthly sales report...
schema ---> date, price, qnt, discount

  step1)
    case class Sales(mon : Int, price:Int,
             qnt :Int, disc: Int)

  step2)
    def toSales(line: String) = {
       val w = line.split(",")
       val mon = w(0).split("/")(0)
       val p = w(1).toInt
       val q = w(2).toInt
       val d = w(3).toInt
       val srec = Sales(mon,p,q,d)
       srec
   }
step3)
 val data = sc.textFile("/user/cloudera/mydata/sales.txt")
step4)
  val strans = data.map(x => toSales(x))
step5)

  val sdf = strans.toDF
  sdf.show

step6)

  sdf.registerTempTable("SalesTrans")

step7) // play with select
  ---> mon, price, qnt, disc

 val res1 = sqlContext.sql("select mon ,
             sum(
             (price -  price*disc/100)*qnt
                ) as tsales from SalesTrans
              group by mon")

 res1.show
 res1.printSchema

-----------------------------------------

 val res2 = res1

 res1.registerTempTable("tab1")
 res2.registerTempTable("tab2")

 val res3 = sqlContext.sql("select l.mon as m1,
     r.mon as m2, l.tsales as t1,
         r.tsales as t2
    from tab1 l join tab2 r
          where (l.mon-r.mon)==1")
//  11 rows.
 res3.registerTempTable("tab3")

------------------------------
 val res4 = sqlContext.sql("select
    m1, m2, t1, t2, ((t2-t1)*100)/t1 as sgrowth
    from tab3")

  res4.show()
------------------------------------------
  json1.json
--------------------------
 {"name":"Ravi","age":25,"city":"Hyd"}
 {"name":"Rani","sex":"F","city":"Del"}
   :
   :
---------------------------------------

    val df = sqlContext.read.json("/user/cloudera/mydata/json1.json")

   df.show
 ------------------------
  name    age    sex    city
 ----------------------------------------
 ravi     25     null hyd
 rani  null   F del
    :
    :
--------------------------------------

 json2.json
 ------------------
 {"name":"Ravi","age":25,
    "wife":{"name":"Rani","age":23},"city":"Hyd"}}
   :
   :

 val df2 = sqlContext.read.json("/../json2.json")

  df2
-----------------------
name     age      wife                  city
Ravi 25    {"name":"rani","age":23}  HYd
    :
---------------------

 df2.registerTempTable("Info")

val df3 = sqlContext.sql("select name,
           wife.name as wname,
           age, wife.age as wage,
          abs(age-wife.age) as diff,
          city from Info")
----------------------------------------
 xml data processing with spark sql.

---spark sql does not have, direct libraries
  for xml processing.
   two ways.
     i) 3 rd party api [ ex: databricks]
    ii) using Hive Integreation.

 2nd is best.

How to integrate Hive with spark .
 ---Using HiveContext.


step1)
   copy hive-site.xml file into,
     /usr/lib/spark/conf directory.

what if , hive-site.xml is not copied into
    conf directory of spark?
--- spark can not understand,
  hive's metastore location [derby/mysql/oracle ....]
    this info is available with hive-site.xml .


 step2)

  create hive Context object

import org.apache.spark.sql.hive.HiveContext
val hc = new HiveContext(sc)

  step3)  access Hive Environment from spark

 hc.sql("create database mydb")
 hc.sql("use mydb")
 hc.sql("create table samp(line string)")
 hc.sql("load data local inpath 'file1'
         into table samp")
 val df = hc.sql("select * from samp")
-------------------------------------
   xml1.xml
----------------------------------
<rec><name>Ravi</name><age>25</age></rec>
<rec><name>Rani</name><sex>F</sex></rec>
 :
 :
------------------------------------------

 hc.sql("create table raw(line string)")
 hc.sql("load data local inpath 'xml1.xml'
   into table raw")
 hc.sql("create table info(name string,
          age int, sex string)")
 hc.sql("insert overwrite table info
    select xpath_string(line,'rec/name'),
       xpath_int(line, 'rec/age'),
      xpath_string(line, 'rec/sex')
      from raw")
----------------------------------------
 xml2.xml
------------
<rec><name><fname>Ravi</fname><lname>kumar</lname><age>24</age><contact><email><personal>ravi@gmail.com</personal><official>ravi@ventech.com</official></email><phone><mobile>12345</mobile><office>123900</office><residence>127845</residence></phone></contact><city>Hyd</city></rec>

hc.sql("create table xraw(line string)")
hc.sql("load data local inpath 'xml2.xml'
   into table xraw")
hc.sql("create table xinfo(fname string ,
   lname string, age int,
    personal_email string,
    official_email string,
    mobile String,
    office_phone string ,
    residence_phone string,
    city string)")

hc.sql("insert overwrite table xinfo
  select
     xpath_string(line,'rec/name/fname'),
     xpath_string(line,'rec/name/lname'),
     xpath_int(line,'rec/age'),
     xpath_string(line,'rec/contact/email/personal'),
  xpath_string(line,'rec/contact/email/official'),
xpath_string(line,'rec/contact/phone/mobile'),
xpath_string(line,'rec/contact/phone/office'),
xpath_string(line,'rec/contact/phone/residence'),
xpath_string(line,'rec/city')
  from xraw")
-------------------------
 xml3.xml
----------------
<tr><cid>101</cid><pr>200</pr><pr>300</pr><pr>300</pr></tr>
<tr><cid>102</cid><pr>400</pr><pr>800</pr></tr>
<tr><cid>101</cid><pr>1000</pr></tr>
--------------------------------

hc.sql("create table sraw")
hc.sql("load data local inpath 'xml3.xml'
    into table sraw")
hc.sql("create table raw2(cid int, pr array<String>)")

hc.sql("insert overwrite table raw2
   select xpath_int(line, 'tr/cid'),
      xpath(line,'tr/pr/text()')
     from sraw")
hc.sql("select * from raw2").show
-------------------------------
cid           pr
101       [100,300,300]
102       [400,800]
101       [1000]

hc.sql("select explode(pr) as price from  raw2").show

    100
    300
    300
    400
    800
   1000

hc.sql("select cid, explode(pr) as price from  raw2").show

----> above is invalid.

hc.sql("create table raw3(cid int, pr int)")
hc.sql("Insert overwrite table raw3
    select name, mypr from raw2
      lateral view explode(pr) p as mypr")

hc.sql("select * from raw3").show

cid     pr
101 200
101 300
101     300
102     400
102     800
101    1000

hc.sql("create table summary(cid int, totbill long)")

hc.sql("insert overwrite table summary
   select cid , sum(pr) from raw3
   group by cid")

--------------------