package com.runnerdev.rdd

import org.apache.spark.SparkConf

import org.apache.spark.SparkContext

 

import org.apache.spark.sql.Dataset

import org.apache.spark.rdd.RDD

import org.apache.spark.sql.SQLContext

 

case class EmployeeData(name: String, age: Long, job: String, salary: Double)

object SparkSqlSchema extends App {

 

  val conf = new SparkConf().setAppName("SparkDataset").setMaster("local")

  // Create a Scala Spark Context.

  val sc = new SparkContext(conf)

 

  // Create a Scala SQLContext .

  val sqlContext = new SQLContext(sc)

  // For implicit conversions from RDDs to DataFrames

  import sqlContext.implicits._

 

  // Create an RDD of Employee objects from a text file, convert it to a Data frame

 

  val employeeDF = sc

    .textFile("/resources/employeeData.txt")

    .map(_.split(","))

    .map(attributes => EmployeeData(attributes(0), attributes(1).trim.toInt, attributes(2), attributes(3).toDouble)).toDF()

  // Register the DataFrame as a temporary view

  employeeDF.registerTempTable("employee")

  println("registerTempTable (employee): ")

  employeeDF.show();

  // SQL statements can be run by using the sql methods provided by Spark

  val sqlQueryDF = sqlContext.sql("SELECT name, age,job,salary FROM employee WHERE age BETWEEN 34 AND 39")

  println("Select query :")

  sqlQueryDF.show()

 

  // The columns of a row in the result can be accessed by field index

  sqlQueryDF.map(empData => "Name: " + empData(0) + " age" + empData(1))

 

  sqlQueryDF.map(empData => "Name: " + empData.getAs[String]("name"))

 

  // row.getValuesMap[T] retrieves multiple columns at once into a Map[String, T]

  sqlQueryDF.map(teenager => teenager.getValuesMap[Any](List("name", "age"))).collect()

} 

 package com.runnerdev.rdd

import org.apache.spark.SparkConf

import org.apache.spark.SparkContext

 

import org.apache.spark.sql.Dataset

import org.apache.spark.rdd.RDD

import org.apache.spark.sql.SQLContext

import org.apache.spark.sql.types.StructType

import org.apache.spark.sql.types.StructField

import org.apache.spark.sql.types.StringType

import org.apache.spark.sql.Row

import org.apache.spark._

 

object SparkSqlSchemaStruct extends App {

 

  val conf = new SparkConf().setAppName("SparkDataset").setMaster("local")

  // Create a Scala Spark Context.

  val sparkContext = new SparkContext(conf)

 

  // Create a Scala SQLContext .

  val sqlContext = new SQLContext(sparkContext)

  // For implicit conversions from RDDs to DataFrames

  import sqlContext.implicits._

 

  // Create an RDD of Employee objects from a text file, convert it to a Data frame

 

  // The schema is encoded in a string

  val schemaString = "name,age,job,salary"

 

  // Generate the schema based on the string of schema

  val fields = schemaString.split(",")

    .map(fieldName => StructField(fieldName, StringType, nullable = true))

  val schema = StructType(fields)

  // Create an RDD

  val employeeRDD = sparkContext.textFile("resources/employeeData.txt")

 

  // Convert records of the RDD (people) to Rows

  val rowRDD = employeeRDD

    .map(_.split(","))

    .map(attributes => Row(attributes(0), attributes(1).trim, attributes(2).trim, attributes(3).trim))

 

  // Apply the schema to the RDD

  val employeeDF = sqlContext.createDataFrame(rowRDD, schema)

 

  // Creates a temporary view using the DataFrame

  employeeDF.registerTempTable("employee")

  employeeDF.show()

  // SQL can be run over a temporary view created using DataFrames

  val results = sqlContext.sql("SELECT name,salary FROM employee")

  results.show()

  // The results of SQL queries are DataFrames and support all the normal RDD operations

  // The columns of a row in the result can be accessed by field index or by field name

 

  results.map(attributes => "salary: " + attributes(1))

 

}