Working with DataFrames

Spark SQL is a Spark module for structured data processing. It provides the programming abstraction called DataFrame (in earlier versions of Spark, it is called SchemaRDD) and also acts as distributed SQL query engine. The capabilities it provides are as follows:

It loads data from a variety of structured sources (for example, JSON, Hive, and Parquet)
It lets you query data using SQL, both inside a Spark program and from external tools that connect to Spark SQL through standard database connectors (JDBC/ODBC), such as BI tools like Tableau.
Spark SQL provides rich integration between SQL and regular Python/Java/Scala code, including the ability to join RDDs and SQL tables, expose custom functions in SQL, and more.

A DataFrame is an RDD of row objects, each representing a record. It is also known as a schema of records. These can be created from external data sources, from results of queries, or from regular RDDs. The created DataFrame can be registered as a temporary table and apply SQLContext.sql or HiveContext.sql to query the table. This recipe shows how to work with DataFrames.

Getting ready

To step through this recipe, you will need a running Spark cluster either in pseudo distributed mode or in one of the distributed modes, that is, standalone, YARN, or Mesos.

How to do it…

Let's see how to create a DataFrame from a JSON file:

         import org.apache.spark.sql._ 
         import org.apache.spark.sql.SQLContext 
         import org.apache.spark.SparkConf 
         import org.apache.spark.SparkContext 
 
         object JSONDataFrame { 
     
         def main(args:Array[String]) 
         { 
         val conf=new SparkConf 
         conf.setMaster("spark://master:7077") 
         conf.setAppName("sql_Sample") 
         val sc=new SparkContext(conf) 
         val sqlcontxt=new SQLContext(sc) 
         val df = sqlContext.read.json("/home/padma/Sparkdev/spark-
         1.6.0/examples/src/main/resources/people.json") 
         df.show 
         df.printSchema 
         df.select("name").show 
         df.select("name","age").show 
         df.select(df("name"),df("age")+4).show 
         df.groupBy("age").count.show 
         df.describe("name,age") } }

Now create a DataFrame from a text file and query on the DataFrame:

        object DataFrames { 
        case class Person(name:String, age:Int) 
        def main(args:Array[String]) 
        { 
        val conf = new SparkConf 
        conf.setMaster("spark://master:7077") 
        conf.setAppName("DataFramesApp") 
        val sc = new SparkContext(conf) 
        val sqlContext = new SQLContext(sc) 
     
        import sqlContext.implicits._ 
        val peopleDf = sc.textFile("/home/padma/Sparkdev/spark-
        1.6.0/examples/src/main/resources/people.txt"). 
        map(line => line.split(",")).map(p => 
        Person(p(0),p(1).trim.toInt)).toDF 
        peopleDf.registerTempTable("people") 
        val teenagers = sqlContext.sql("select name, age from people 
        where age >=13 AND name in(select name from people where age=
        30)") 
        teenagers.map(t => "Name: " + t(0)).collect().foreach(println) 
          }
        }

Here is the code snippet to show how to create a DataFrame from a parquet file:

        val sc = new SparkContext(conf) 
        val sqlContext = new SQLContext(sc) 
        import sqlContext.implicits._ 
     
        val df1 = sc.makeRDD(1 to 5).map(i =>  
        (i,i*2)).toDF("single","double") 
        df1.write.parquet("/home/padma/Sparkdev/SparkApp/
        test_table/key=1") 
     
        val df2 = sc.makeRDD(6 to 10).map(i => 
        (i,i*4)).toDF("single","triple") 
        df2.write.parquet("/home/padma/Sparkdev/SparkApp/
        test_table/key=2") 
     
        val df3 = sqlContext.read.parquet("/home/padma/Sparkdev/
        SparkApp/test_table") 
        df3.show

How it works…

Initially, the JSON file is read, which is the DataFrame, and the API such as show(), printSchema(), select(), or groupBy() can be invoked on the data frame. In the second code snippet, an RDD is created from the text file and the fields are mapped to the case class structure Person and the RDD is converted to a data frame using toDF. This data frame peopleDF is converted to a table using registerTempTable() whose table name is people. Now this table people can be queried using SQLContext.sql.

The final code snippet shows how to write a data frame as a parquet file using df1.write.parquet() and the parquet file is read using sqlContext.read.parquet().

There's more…

Spark SQL in addition provides HiveContext, using which we can access Hive tables, UDFS, SerDes, and also HiveQL. There are ways to create DataFrames by converting an RDD to a DataFrame or creating them programmatically. The different data sources, such as JSON, Parquet, and Avro, can be handled and there is provision to directly run sql queries on the files. Also, data from other databases can be read using JDBC. In Spark 1.6.0, a new feature known as Dataset is introduced, which provides the benefits of Spark SQL's optimized execution engine over RDDs.

Apache Spark for Data Science Cookbook

By : Nagamallikarjuna Inelu, Chitturi

Apache Spark for Data Science Cookbook

By: Nagamallikarjuna Inelu, Chitturi

Overview of this book

Working with DataFrames

Getting ready

How to do it…

How it works…

There's more…

See also

Apache Spark for Data Science Cookbook

By : Nagamallikarjuna Inelu, Chitturi

Apache Spark for Data Science Cookbook

By: Nagamallikarjuna Inelu, Chitturi

Overview of this book

Working with DataFrames

Getting ready

How to do it…

How it works…

There's more…

See also

Confirmation

Buy this book with your credits?

Submit Your Feedback

Create a Free Account To Continue Reading

Sign in to activate your 7-day free access