How to Perform Faster Read/write Operation From Spark.

Apache Spark is a lightning-fast cluster computing framework designed for fast ciphering. With the advent of real-time processing framework in the Big Data Ecosystem, companies are using Apache Spark rigorously in their solutions. Spark SQL is a new module in Spark which integrates relational processing with Spark'south functional programming API. It supports querying data either via SQL or via the Hive Query Language. Through this blog, I will introduce you to this new exciting domain of Spark SQL.

The following provides the storyline for the weblog:

• What is Spark SQL?
• Why is Spark SQL used?
• How does Spark SQL piece of work?
• Spark SQL Libraries
• Features of Spark SQL
• Querying using Spark SQL
• Calculation Schema to RDDs
• RDDs as Relations
• Caching Tables In-Memory

What is Spark SQL?

Spark SQL integrates relational processing with Spark's functional programming. Information technology provides support for various data sources and makes it possible to weave SQL queries with code transformations thus resulting in a very powerful tool.

Why is Spark SQL used?

Spark SQL originated as Apache Hive to run on top of Spark and is at present integrated with the Spark stack. Apache Hive had sure limitations equally mentioned beneath. Spark SQL was built to overcome these drawbacks and supersede Apache Hive.

Is Spark SQL faster than Hive?

Spark SQL is faster than Hive when it comes to processing speed. Below I have listed down a few limitations of Hive over Spark SQL.

Limitations With Hive:

• Hive launches MapReduce jobs internally for executing the ad-hoc queries. MapReduce lags in the functioning when it comes to the assay of medium-sized datasets (10 to 200 GB).
• Hive has no resume capability. This means that if the processing dies in the middle of a workflow, you cannot resume from where information technology got stuck.
• Hive cannot drop encrypted databases in cascade when the trash is enabled and leads to an execution error. To overcome this, users accept to use the Purge option to skip trash instead of driblet.

These drawbacks gave way to the birth of Spark SQL. Merely the question which still pertains in near of our minds is,

Is Spark SQL a database?

Spark SQL is non a database just a module that is used for structured information processing. Information technology majorly works on DataFrames which are the programming abstraction and usually act as a distributed SQL query engine.

How does Spark SQL piece of work?

Let u.s.a. explore, what Spark SQL has to offer. Spark SQL blurs the line between RDD and relational table. Information technology offers much tighter integration between relational and procedural processing, through declarative DataFrame APIs which integrates with Spark code. It also provides higher optimization. DataFrame API and Datasets API are the ways to interact with Spark SQL.

With Spark SQL, Apache Spark is accessible to more users and improves optimization for the electric current ones. Spark SQL provides DataFrame APIs which perform relational operations on both external information sources and Spark's built-in distributed collections. It introduces an extensible optimizer chosen Catalyst as information technology helps in supporting a wide range of data sources and algorithms in Big-data.

Spark runs on both Windows and UNIX-like systems (due east.k. Linux, Microsoft, Mac Bone). Information technology is easy to run locally on one car — all you demand is to have java installed on your system PATH, or the JAVA_HOME environs variable pointing to a Java installation.

Effigy: Architecture of Spark SQL.

Spark SQL Libraries

Spark SQL has the following 4 libraries which are used to interact with relational and procedural processing:

ane. Information Source API (Application Programming Interface):

This is a universal API for loading and storing structured data.

• Information technology has built-in support for Hive, Avro, JSON, JDBC, Parquet, etc.
• Supports 3rd-political party integration through Spark packages
• Support for smart sources.
• It is a Data Abstraction and Domain Specific Linguistic communication (DSL) applicable to structure and semi-structured information.
• DataFrame API is a distributed drove of information in the form of named column and row.
• It is lazily evaluated like Apache Spark Transformations and can be accessed through SQL Context and Hive Context.
• Information technology processes the data in the size of Kilobytes to Petabytes on a single-node cluster to multi-node clusters.
• Supports different data formats (Avro, CSV, Elastic Search, and Cassandra) and storage systems (HDFS, HIVE Tables, MySQL, etc.).
• Can be easily integrated with all Big Information tools and frameworks via Spark-Core.
• Provides API for Python, Java, Scala, and R Programming.

2. DataFrame API:

A DataFrame is a distributed drove of data organized into named columns. Itis equivalent to a relational table in SQL  used for storing data into tables.

iii. SQL Interpreter And Optimizer:

SQL Interpreter and Optimizer is based on functional programming synthetic in Scala.

• Information technology is the newest and most technically evolved component of SparkSQL.
• It provides a general framework for transforming copse, which is used to perform assay/evaluation, optimization, planning, and run fourth dimension code spawning.
• This supports cost-based optimization (run time and resource utilization are termed as cost) and rule-based optimization, making queries run much faster than their RDD (Resilient Distributed Dataset) counterparts.

     east.m. Catalyst is a modular library that is made as a rule-based system. Each dominion in the framework focuses on distinct optimization.

four. SQL Service:

SQL Service is the entry point for working forth with structured data in Spark. Information technology allows the creation of DataFrame objects likewise as the execution of SQL queries.

Features Of Spark SQL

The following are the features of Spark SQL:

1. Integration With Spark

   Spark SQL queries are integrated with Spark programs. Spark SQL allows us to query structured information inside Spark programs, using SQL or a DataFrame API which can be used in Coffee, Scala, Python and R. To run the streaming computation, developers merely write a batch ciphering against the DataFrame / Dataset API, and Spark automatically increments the computation to run it in a streaming fashion. This powerful pattern means that developers don't have to manually manage country, failures, or keeping the application in sync with batch jobs. Instead, the streaming job always gives the same answer every bit a batch job on the aforementioned information.

2. Uniform Data Admission

   DataFrames and SQL support a mutual style to admission a diversity of data sources, similar Hive, Avro, Parquet, ORC, JSON, and JDBC. This joins the data across these sources. This is very helpful to accommodate all the existing users into Spark SQL.
   

3. Hive Compatibility

   Spark SQL runs unmodified Hive queries on electric current data. It rewrites the Hive front end-end and meta store, allowing full compatibility with current Hive data, queries, and UDFs.
   

4. Standard Connectivity

   The connection is through JDBC or ODBC. JDBC and ODBC are the industry norms for connectivity for business intelligence tools.

5. Performance And Scalability

   Spark SQL incorporates a price-based optimizer, code generation, and columnar storage to make queries agile alongside calculating thousands of nodes using the Spark engine, which provides full mid-query fault tolerance. The interfaces provided past Spark SQL provide Spark with more than information about the structure of both the data and the ciphering being performed. Internally, Spark SQL uses this extra information to perform extra optimization. Spark SQL can straight read from multiple sources (files, HDFS, JSON/Parquet files, existing RDDs, Hive, etc.). It ensures the fast execution of existing Hive queries.
   The image beneath depicts the performance of Spark SQL when compared to Hadoop. Spark SQL executes upwards to 100x times faster than Hadoop.

Effigy: Runtime of Spark SQL vs Hadoop. Spark SQL is faster
Source: Cloudera Apache Spark Web log

1. User-Defined Functions
   Spark SQL has language integrated User-Divers Functions (UDFs). UDF is a feature of Spark SQL to define new Column-based functions that extend the vocabulary of Spark SQL's DSL for transforming Datasets. UDFs are black boxes in their execution.

   The example below defines a UDF to catechumen a given text to upper instance.

   Code explanation:
   1. Creating a dataset "hi world"
   ii. Defining a role 'upper' which converts a string into upper instance.
   3. Nosotros now import the 'udf' bundle into Spark.
   4. Defining our UDF, 'upperUDF' and importing our function 'upper'.
   5. Displaying the results of our User Defined Function in a new column 'upper'.

   val dataset = Seq((0, "hello"),(1, "world")).toDF("id","text") val upper: String => Cord =_.toUpperCase import org.apache.spark.sql.functions.udf val upperUDF = udf(upper) dataset.withColumn("upper", upperUDF('text)).show            

Figure:Demonstration of a User Defined Function, upperUDF

Code caption:
1. We now register our function every bit 'myUpper'
two. Cataloging our UDF among the other functions.

spark.udf.register("myUpper", (input:String) => input.toUpperCase) spark.itemize.listFunctions.filter('name like "%upper%").show(false)        

  Effigy: Results of the User Divers Role, upperUDF

Querying Using Spark SQL

We will now start querying using Spark SQL. Note that the actual SQL queries are similar to the ones used in popular SQL clients.

Starting the Spark Shell. Go to the Spark directory and execute ./bin/spark-shell in the terminal to being the Spark Beat out.

For the querying examples shown in the web log, we will be using two files, 'employee.txt' and 'employee.json'. The images below evidence the content of both the files. Both these files are stored at 'examples/src/chief/scala/org/apache/spark/examples/sql/SparkSQLExample.scala' inside the folder containing the Spark installation (~/Downloads/spark-2.0.2-bin-hadoop2.7). So, all of you lot who are executing the queries, place them in this directory or gear up the path to your files in the lines of code below.

Figure:Contents of employee.txt

Figure:Contents of employee.json

Lawmaking explanation:
ane. We outset import a Spark Session into Apache Spark.
2. Creating a Spark Session 'spark' using the 'builder()' part.
three. Importing the Implicts class into our 'spark' Session.
4. We now create a DataFrame 'df' and import data from the 'employee.json' file.
5. Displaying the DataFrame 'df'. The result is a table of 5 rows of ages and names from our 'employee.json' file.

import org.apache.spark.sql.SparkSession val spark = SparkSession.builder().appName("Spark SQL bones example").config("spark.some.config.option", "some-value").getOrCreate() import spark.implicits._ val df = spark.read.json("examples/src/principal/resources/employee.json") df.show()        

Figure:Starting a Spark Session and displaying DataFrame of employee.json

Code explanation:
1. Importing the Implicts course into our 'spark' Session.
2. Press the schema of our 'df' DataFrame.
3. Displaying the names of all our records from 'df' DataFrame.

import spark.implicits._ df.printSchema() df.select("name").show()        

Figure:Schema of a DataFrame

Code explanation:
1. Displaying the DataFrame later incrementing everyone'southward age by two years.
ii. Nosotros filter all the employees above age 30 and display the outcome.

df.select($"proper name", $"age" + ii).show() df.filter($"age" > xxx).show()        

Effigy:Basic SQL operations on employee.json

Code caption:
1. Counting the number of people with the same ages. We utilise the 'groupBy' part for the aforementioned.
2. Creating a temporary view 'employee' of our 'df' DataFrame.
3. Perform a 'select' operation on our 'employee' view to brandish the table into 'sqlDF'.
4. Displaying the results of 'sqlDF'.

df.groupBy("age").count().show() df.createOrReplaceTempView("employee") val sqlDF = spark.sql("SELECT * FROM employee") sqlDF.show()        

Effigy: SQL operations on employee.json

Creating Datasets

After understanding DataFrames, let the states now motility on to Dataset API. The below code creates a Dataset class in SparkSQL.

Code explanation:
1. Creating a grade 'Employee' to store name and age of an employee.
2. Assigning a Dataset 'caseClassDS' to shop the record of Andrew.
3. Displaying the Dataset 'caseClassDS'.
4. Creating a primitive Dataset to demonstrate mapping of DataFrames into Datasets.
5. Assigning the above sequence into an array.

case grade Employee(name: Cord, age: Long) val caseClassDS = Seq(Employee("Andrew", 55)).toDS() caseClassDS.prove() val primitiveDS = Seq(1, 2, 3).toDS ()primitiveDS.map(_ + 1).collect()        

Figure:Creating a Dataset

Code explanation:
1. Setting the path to our JSON file 'employee.json'.
ii. Creating a Dataset and from the file.
three. Displaying the contents of 'employeeDS' Dataset.

val path = "examples/src/master/resources/employee.json" val employeeDS = spark.read.json(path).as[Employee] employeeDS.show()        

Effigy:Creating a Dataset from a JSON file

Adding Schema To RDDs

Spark introduces the concept of an RDD (Resilient Distributed Dataset), an immutable fault-tolerant, distributed collection of objects that tin exist operated on in parallel. An RDD tin can incorporate whatsoever type of object and is created past loading an external dataset or distributing a collection from the driver plan.

Schema RDD is a RDD where you can run SQL on. It is more than SQL. It is a unified interface for structured data.

Code caption:
1. Importing Expression Encoder for RDDs. RDDs are similar to Datasets only use encoders for serialization.
2. Importing Encoder library into the shell.
3. Importing the Implicts form into our 'spark' Session.
four. Creating an 'employeeDF' DataFrame from 'employee.txt' and mapping the columns based on the delimiter comma ',' into a temporary view 'employee'.
5. Creating the temporary view 'employee'.
6. Defining a DataFrame 'youngstersDF' which will contain all the employees between the ages of 18 and xxx.
7. Mapping the names from the RDD into 'youngstersDF' to display the names of youngsters.

import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder import org.apache.spark.sql.Encoder import spark.implicits._ val employeeDF = spark.sparkContext.textFile("examples/src/main/resources/employee.txt").map(_.divide(",")).map(attributes => Employee(attributes(0), attributes(1).trim.toInt)).toDF() employeeDF.createOrReplaceTempView("employee") val youngstersDF = spark.sql("SELECT proper noun, historic period FROM employee WHERE age BETWEEN 18 AND 30") youngstersDF.map(youngster => "Proper name: " + youngster(0)).show()        

Figure:Creating a DataFrame for transformations

Code explanation:
ane. Converting the mapped names into string for transformations.
2. Using the mapEncoder from Implicits form to map the names to the ages.
3. Mapping the names to the ages of our 'youngstersDF' DataFrame. The result is an array with names mapped to their respective ages.

youngstersDF.map(youngster => "Proper noun: " + youngster.getAs[String]("name")).show() implicit val mapEncoder = org.apache.spark.sql.Encoders.kryo[Map[Cord, Whatever]] youngstersDF.map(youngster => youngster.getValuesMap[Any](List("name", "age"))).collect()        

Figure:Mapping using DataFrames

RDDs support two types of operations:

• Transformations: These are the operations (such every bit map, filter, join, union, and so on) performed on an RDD which yield a new RDD containing the result.
• Actions: These are operations (such as reduce, count, commencement, and so on) that render a value after running a computation on an RDD.

Transformations in Spark are "lazy", pregnant that they do not compute their results right abroad. Instead, they just "remember" the functioning to be performed and the dataset (eastward.g., file) to which the operation is to exist performed. The transformations are computed merely when an action is called and the effect is returned to the driver program and stored equally Directed Acyclic Graphs (DAG). This design enables Spark to run more efficiently. For example, if a big file was transformed in diverse means and passed to outset action, Spark would only procedure and return the event for the kickoff line, rather than do the work for the entire file.

Figure:Ecosystem of Schema RDD in Spark SQL

By default, each transformed RDD may be recomputed each time you run an activeness on it. However, you may as well persist an RDD in retention using the persist or cache method, in which case Spark volition keep the elements effectually on the cluster for much faster access the next time y'all query information technology.

RDDs As Relations

Resilient Distributed Datasets (RDDs) are distributed memory abstraction which lets programmers perform in-memory computations on big clusters in a fault tolerant manner. RDDs tin can be created from any data source. Eg: Scala collection, local file arrangement, Hadoop, Amazon S3, HBase Table, etc.

Specifying Schema

Code explanation:
1. Importing the 'types' class into the Spark Shell.
2. Importing 'Row' class into the Spark Shell. Row is used in mapping RDD Schema.
3. Creating a RDD 'employeeRDD' from the text file 'employee.txt'.
4. Defining the schema every bit "name age". This is used to map the columns of the RDD.
5. Defining 'fields' RDD which will exist the output later mapping the 'employeeRDD' to the schema 'schemaString'.
half-dozen. Obtaining the type of 'fields' RDD into 'schema'.

import org.apache.spark.sql.types._ import org.apache.spark.sql.Row val employeeRDD = spark.sparkContext.textFile("examples/src/main/resources/employee.txt") val schemaString = "name age" val fields = schemaString.divide(" ").map(fieldName => StructField(fieldName, StringType, nullable = truthful)) val schema = StructType(fields)        

Effigy:Specifying Schema for RDD transformation

Lawmaking caption:
ane. Nosotros now create a RDD called 'rowRDD' and transform the 'employeeRDD' using the 'map' part into 'rowRDD'.
2. We ascertain a DataFrame 'employeeDF' and store the RDD schema into information technology.
3. Creating a temporary view of 'employeeDF' into 'employee'.
4. Performing the SQL performance on 'employee' to display the contents of employee.
five. Displaying the names of the previous operation from the 'employee' view.

val rowRDD = employeeRDD.map(_.carve up(",")).map(attributes => Row(attributes(0), attributes(1).trim)) val employeeDF = spark.createDataFrame(rowRDD, schema) employeeDF.createOrReplaceTempView("employee") val results = spark.sql("SELECT name FROM employee") results.map(attributes => "Proper noun: " + attributes(0)).show()        

Figure:Result of RDD transformation

Fifty-fifty though RDDs are defined, they don't contain any data. The computation to create the information in a RDD is only washed when the information is referenced. e.thou. Caching results or writing out the RDD.

Caching Tables In-Memory

Spark SQL caches tables using an in-memory columnar format:

1. Scan only required columns
2. Fewer allocated objects
3. Automatically selects best comparison

Loading Data Programmatically

The below code will read employee.json file and create a DataFrame. We will then use information technology to create a Parquet file.

Code explanation:
1. Importing Implicits class into the vanquish.
two. Creating an 'employeeDF' DataFrame from our 'employee.json' file.

import spark.implicits._ val employeeDF = spark.read.json("examples/src/principal/resources/employee.json")        

Figure:Loading a JSON file into DataFrame

Code explanation:
i. Creating a 'parquetFile' temporary view of our DataFrame.
two. Selecting the names of people between the ages of xviii and xxx from our Parquet file.
3. Displaying the result of the Spark SQL operation.

employeeDF.write.parquet("employee.parquet") val parquetFileDF = spark.read.parquet("employee.parquet") parquetFileDF.createOrReplaceTempView("parquetFile") val namesDF = spark.sql("SELECT proper noun FROM parquetFile WHERE age BETWEEN 18 AND 30") namesDF.map(attributes => "Name: " + attributes(0)).show()        

Figure:Displaying results from a Parquet DataFrame

JSON Datasets

We will at present work on JSON information. Every bit Spark SQL supports JSON dataset, we create a DataFrame of employee.json. The schema of this DataFrame tin can be seen below. We so define a Youngster DataFrame and add all the employees between the ages of 18 and 30.

Lawmaking explanation:
1. Setting to path to our 'employee.json' file.
two. Creating a DataFrame 'employeeDF' from our JSON file.
3. Printing the schema of 'employeeDF'.
four. Creating a temporary view of the DataFrame into 'employee'.
5. Defining a DataFrame 'youngsterNamesDF' which stores the names of all the employees between the ages of xviii and 30 present in 'employee'.
six. Displaying the contents of our DataFrame.

val path = "examples/src/main/resources/employee.json" val employeeDF = spark.read.json(path) employeeDF.printSchema() employeeDF.createOrReplaceTempView("employee") val youngsterNamesDF = spark.sql("SELECT name FROM employee WHERE age Between xviii AND 30") youngsterNamesDF.show()        

Figure:Operations on JSON Datasets

Code explanation:
i. Creating a RDD 'otherEmployeeRDD' which will store the content of employee George from New Delhi, Delhi.
2. Assigning the contents of 'otherEmployeeRDD' into 'otherEmployee'.
3. Displaying the contents of 'otherEmployee'.

val otherEmployeeRDD = spark.sparkContext.makeRDD("""{"name":"George","accost":{"urban center":"New Delhi","state":"Delhi"}}""" :: Nix) val otherEmployee = spark.read.json(otherEmployeeRDD) otherEmployee.show()        

Effigy:RDD transformations on JSON Dataset

Hive Tables

We perform a Spark example using Hive tables.

Code caption:
1. Importing 'Row' class into the Spark Shell. Row is used in mapping RDD Schema.
2. Importing Spark Session into the shell.
3. Creating a class 'Record' with attributes Int and String.
4. Setting the location of 'warehouseLocation' to Spark warehouse.
5. We at present build a Spark Session 'spark' to demonstrate Hive example in Spark SQL.
6. Importing Implicits class into the shell.
7. Importing SQL library into the Spark Vanquish.
8. Creating a tabular array 'src' with columns to store fundamental and value.

import org.apache.spark.sql.Row import org.apache.spark.sql.SparkSession case form Record(fundamental: Int, value: String) val warehouseLocation = "spark-warehouse" val spark = SparkSession.builder().appName("Spark Hive Example").config("spark.sql.warehouse.dir", warehouseLocation).enableHiveSupport().getOrCreate() import spark.implicits._ import spark.sql sql("CREATE Table IF Not EXISTS src (key INT, value String)")        

Figure:Building a Session for Hive

Lawmaking caption:
1. We now load the data from the examples present in Spark directory into our tabular array 'src'.
two. The contents of 'src' is displayed beneath.

sql("LOAD Information LOCAL INPATH 'examples/src/main/resources/kv1.txt' INTO Table src") sql("SELECT * FROM src").evidence()        

Figure:Choice using Hive tables

Code explanation:
1. Nosotros perform the 'count' performance to select the number of keys in 'src' table.
2. We at present select all the records with 'primal' value less than 10 and shop it in the 'sqlDF' DataFrame.
three. Creating a Dataset 'stringDS' from 'sqlDF'.
iv. Displaying the contents of 'stringDS' Dataset.

sql("SELECT COUNT(*) FROM src").bear witness() val sqlDF = sql("SELECT central, value FROM src WHERE cardinal < 10 ORDER By fundamental") val stringsDS = sqlDF.map {case Row(primal: Int, value: String) => s"Key: $key, Value: $value"} stringsDS.show()        

Figure: Creating DataFrames from Hive tables

Code explanation:
1. We create a DataFrame 'recordsDF' and shop all the records with key values 1 to 100.
2. Create a temporary view 'records' of 'recordsDF' DataFrame.
3. Displaying the contents of the join of tables 'records' and 'src' with 'key' as the primary fundamental.

val recordsDF = spark.createDataFrame((1 to 100).map(i => Record(i, southward"val_$i"))) recordsDF.createOrReplaceTempView("records") sql("SELECT * FROM records r JOIN src southward ON r.fundamental = s.key").evidence()        

Effigy: Recording the results of Hive operations

So this concludes our web log. I hope you enjoyed reading this blog and found information technology informative. By now, you lot must take caused a audio understanding of what Spark SQL is. The hands-on examples will give you lot the required confidence to work on any future projects you come across in Spark SQL. Practice is the key to mastering any subject and I hope this web log has created plenty interest in you to explore learning further on Spark SQL.

Got a question for united states of america? Please mention it in the comments section and we will get back to y'all at the earliest.

If you lot wish to acquire Spark and build a career in domain of Spark and build expertise to perform big-scale Data Processing using RDD, Spark Streaming, SparkSQL, MLlib, GraphX and Scala with Real Life employ-cases, check out our interactive, live-online Apache Spark Certification Training here, that comes with 24*7 support to guide yous throughout your learning period.

schrockmusidn.blogspot.com

Source: https://www.edureka.co/blog/spark-sql-tutorial/

Share this post