Basic Query Example

Let’s walk through a few examples of queries on a data set of US flight delays with date, delay, distance, origin, and destination. ⁴ In our example, we will load a CSV file with over a million records. Using a schema for the CSV, we read data into a DataFrame and register the DataFrame as a temporary view (more on temporary views shortly) so we can query it with SQL.

For query examples, see all the code snippets in Examples 4-1 through 4-5, and for the entire example notebook in Python and Scala, see the code in the Learning Spark v2 GitHub⁵ repo. All these examples of SQL queries offer you a taste in how to use SQL in your Spark application using the spark.sql programmatic interface.⁶ Similar to the DataFrame API, this interface lets you query structured data in your Spark application.

Normally, in a standalone Spark application, you will have to create a SparkSession instance, as shown in the example below. However, in a Spark shell or Databricks notebook, the SparkSession is created for you, named appropriately as spark.

val csvFile = …         // path to your dataset; 
val spark = SparkSession
         .builder
         .appName("SparkSQLExampleApp")
.getOrCreate()
// define our schema as before
val schema = StructType(StructField("date", StringType, false),
         StructField("delay", IntegerType, false),
         StructField("distance", IntegerType, false),
StructField("origin", StringType, false),
StructField("destination", StringType, false))
// read and create a temporary view
spark.read.schema(schema).csv(csvFile).createOrReplaceTempView("us_delay_flights_tbl")

     csvFile = …             # path to your dataset
       # create a SparkSession
       spark = (SparkSession
                .builder
                .appName("SparkSQLExampleApp")
                .getOrCreate())
      # define a schema
      schema = StructType([
           StructField("date", StringType(), False),
           StructField("delay", IntegerType(), False),
           StructField("distance", IntegerType(), False),
           StructField("origin", StringType(), False),
           StructField("destination", StringType(), False)])
    # read the data and create a temporary view spark.read.schema(schema).csv(csv_file).createOrReplaceTempView("us_delay_flights_tbl")

    // in Scala
    val schema = “date STRING, delay INT, distance INT, origin STRING”, destination STRING”
    # In Python
    schema = “`date` STRING, `delay`INT, `distance` INT, `origin` STRING, `destination` STRING”

Now that we have a temporary view, we can issue SQL queries using Spark SQL. These SQL queries are no different than those you may issue against a SQL table, say in MySQL or PostgreSQL database. The point here is to show that Spark SQL offers ANSI 2003 compliant SQL interface—and the interoperability between SQL and DataFrames.

So let’s have a go at some queries against our US departures flight data set mentioned above. The data set has five columns:

D ate

string e.g.,02190925, when converted maps to 02-19 09:25 am

D elay

in minutes from the scheduled time; a negative number denotes late arrival of the flight to the airport of origin, and positive denotes a late departure?

D istance

in miles from the origin airport to the destination airport

origin

origin IATA airport code

destination

destination IATA airport code

spark.sql(“SELECT distance, origin, destination 
FROM us_delay_flights_tbl WHERE distance > 1000 
ORDER BY distance DESC”).show(10)

This query produces a result that looks as follows, showing the top 10 farthest in distance from Honolulu to New York.

+--------+------+-----------+
|distance|origin|destination|
+--------+------+-----------+
|4330    |HNL   |JFK        |
|4330    |HNL   |JFK        |
|4330    |HNL   |JFK        |
|4330    |HNL   |JFK        |
|4330    |HNL   |JFK        |
|4330    |HNL   |JFK        |
|4330    |HNL   |JFK        |
|4330    |HNL   |JFK        |
|4330    |HNL   |JFK        |
|4330    |HNL   |JFK        |
+--------+------+-----------+
only showing top 10 rows

spark.sql("SELECT date, delay, origin, destination 
FROM us_delay_flights_tbl 
WHERE delay> 120 AND ORIGIN = 'SFO' AND DESTINATION = 'ORD' 
ORDER by delay DESC").show(10)

+--------+-----+------+-----------+
|date    |delay|origin|destination|
+--------+-----+------+-----------+
|02190925|1638 |SFO   |ORD        |
|01031755|396  |SFO   |ORD        |
|01022330|326  |SFO   |ORD        |
|01051205|320  |SFO   |ORD        |
|01190925|297  |SFO   |ORD        |
|02171115|296  |SFO   |ORD        |
|01071040|279  |SFO   |ORD        |
|01051550|274  |SFO   |ORD        |
|03120730|266  |SFO   |ORD        |
|01261104|258  |SFO   |ORD        |
+--------+-----+------+-----------+
only showing top 10 rows

It seems there were many significantly delayed flights between San Francisco and Chicago on different dates. (As an exercise, convert date column into readable format and ascertain what days or months were these delays? Were they related to winter months or holidays?)

Let’s try a more complicated query where we use the CASE clause in SQL. We want to label all US flights originating from airports with high, medium, low, no delays, regardless of destinations.

spark.sql("SELECT delay, origin, destination,
              CASE
                  WHEN delay > 360 THEN 'Very Long Delays'
                  WHEN delay > 120 AND delay < 360 THEN  'Long Delays'
                  WHEN delay > 60 AND delay < 120 THEN  'Short Delays'
                  WHEN delay > 0 and delay < 60  THEN   'Tolerable  Delays'
                  WHEN delay = 0 THEN 'No Delays'
                  ELSE 'Early'
               END AS Flight_Delays
               FROM us_delay_flights_tbl
               ORDER BY origin, delay DESC").show(10)

+-----+------+-----------+-------------+
|delay|origin|destination|Flight_Delays|
+-----+------+-----------+-------------+
|333  |ABE   |ATL        |Long Delays  |
|305  |ABE   |ATL        |Long Delays  |
|275  |ABE   |ATL        |Long Delays  |
|257  |ABE   |ATL        |Long Delays  |
|247  |ABE   |DTW        |Long Delays  |
|247  |ABE   |ATL        |Long Delays  |
|219  |ABE   |ORD        |Long Delays  |
|211  |ABE   |ATL        |Long Delays  |
|197  |ABE   |DTW        |Long Delays  |
|192  |ABE   |ORD        |Long Delays  |
+-----+------+-----------+-------------+
only showing top 10 rows

As with DataFrames and Datasets APIs so with the spark.sql interface, you can conduct common data analysis we explored in the previous chapter; both the DataFrame as well as the spark.sql query computation undergo an identical journey in the Spark SQL engine (see Figure 3-2 in chapter 3), giving you the same results.

from pyspark.sql.functions import col, desc

(df.select("distance", "origin", "destination")
.where(col("distance") > 1000)
.orderBy(desc("distance")).show())

(df.select("distance", "origin", "destination")
   .where("distance > 1000")
   .orderBy("distance", ascending=False).show(10)

+--------+------+-----------+
|distance|origin|destination|
+--------+------+-----------+
|4330    |HNL   |JFK        |
|4330    |HNL   |JFK        |
|4330    |HNL   |JFK        |
|4330    |HNL   |JFK        |
|4330    |HNL   |JFK        |
|4330    |HNL   |JFK        |
|4330    |HNL   |JFK        |
|4330    |HNL   |JFK        |
|4330    |HNL   |JFK        |
|4330    |HNL   |JFK        |
+--------+------+-----------+
only showing top 10 rows

As you can see from the above queries, using the Spark SQL interface to query data feels like you are writing a regular SQL query to a relational database table. Although the queries are in SQL, you can feel the similarity in readability and semantics to DataFrames API operations, which you encountered in chapter 3—and will explore more in the next chapter.

For you to be able to query structured data, as shown in the above examples, Spark manages all the complexities of creating and managing views and tables, both in memory and on disk, which leads us to discuss how tables and views are created and managed.

Unmanaged vs Managed Tables

Spark allows you to create two types of tables: unmanaged and managed. For managed tables, Spark manages both the metadata and data on the file store. This could be local file system, HDFS, or an object store such as Amazon S3 or Azure Blob.

Because Spark manages the table, a SQL command such as DROP table_name SQL deletes both the metadata and data. Whereas in an unmanaged table, the same command will delete only the metadata, not the actual data. For an unmanaged table, Spark only manages the metadata, while you can manage the data in an external data source such as Cassandra, JDBC or other.⁷ We will look at some examples of how to create managed and unmanaged tables below.

Creating SQL Databases and Tables

Tables reside within a Database. By default, Spark creates tables under a default database. To create your own Database name, you can issue a SQL command from your Spark application or notebook. Using the US delay flights dataset above, let’s create both a managed and unmanaged table.

spark.sql(“CREATE DATABASE learn_spark_db”)
spark.sql(“USE learn_spark_db”)

Any subsequent SQL commands for creating tables for the duration of a Spark application will reside under the database name learn_spark_db.

spark.sql(“CREATE TABLE us_delay_flights_tbl (date STRING, delay INT, distance INT, origin STRING, destination STRING”)

In Python DataFrame API use:

# path to our US flight delay CSV file 
csv_file = . . .
# schema as defined in the above example
schema = “date STRING, delay INT, distance INT, origin STRING, destination STRING”
flights_df = spark.read.csv(csv_file, schema=schema)
flights_df.saveAsTable(“us_delay_flights_tbl”)

spark.sql(“CREATE TABLE us_delay_flights_tbl(date STRING, delay INT, 
  distance INT, origin STRING, destination STRING) 
  USING json OPTIONS (PATH ‘path_to_us_flights_json_file’)”)

# From SQL 
CREATE TABLE us_delay_flights_tbl(date STRING, delay INT, 
distance INT, origin STRING, destination STRING) 
USING json OPTIONS (path "path_to_us_flights_json_file")

dataframe.write.option('path',"/tmp/data/us_flights_delay").saveAsTable("us_delay_flights_tbl")

Creating Views

In addition to creating tables, Spark can create views on top of existing tables. Views can be global (visible across all SparkSessions on a given cluster) or temporary (visible only to a single SparkSession), and they disappear after your Spark application or notebook terminates.

Like creating tables within a database, creating views has a similar syntax.⁹ Once you create a view, you can query it as you would a table. The difference between a view and table is that views allow you organize data with different logical lens or slices of the same table. It’s like creating multiple DataFrames with different transformational queries from the original DataFrame.

You can also create a view from an existing Table using SQL. For example, if you wish to work only on a subset of US flights with origin airports in New York (JFK) and San Francisco (SFO), the following queries will create a slice of the table as a view.

In SQL:

CREATE OR REPLACE TEMP VIEW us_origin_airport_SFO_global_tmp_view AS
  SELECT date, delay, origin, destination from us_delay_flights_tbl WHERE origin = ‘SFO’

CREATE OR REPLACE GLOBAL TEMP VIEW us_origin_airport_JFK_tmp_view AS
  SELECT date, delay, origin, destination from us_delay_flights_tbl WHERE origin = ‘JFK’

Alternatively, for a given a DataFrame, you can call its two methods to create temporary or global temporary views (we will discuss the difference between them shortly):

# In Python
df_sfo =  spark.sql(“SELECT date, delay, origin, destination  from us_delay_flights_tbl WHERE origin = ‘SFO’”)
df_jfk = spark.sql(“SELECT date, delay, origin, destination from us_delay_flights_tbl WHERE origin = ‘JFK’”)

# create a temporary and global temporary view
df_sfo.createOrReplaceGlobalTempView(“us_origin_airport_SFO_global_tmp_view”)
df_jfk.createOrReplaceTempView(“us_origin_airport_JFK_tmp_view”)

Let’s issue queries against both the temporary and global view. Keep in mind that, while accessing a global temporary view, you must use the prefix “global_temp.<view_or_table_name” because Spark creates global temporary view in a global temporary database called global_temp.

Once created, you can issue queries against these temporary views as you would against tables–and drop a view as you would a table.

In SQL use: 
SELECT * FROM global_temp.us_origin_airport_SFO_global_tmp_view

By contrast, you can access the normal temporary view without the global_temp prefix.

In SQL use: 
SELECT * FROM us_origin_airport_JFK_tmp_view

And to drop the view or table, simply use:

In In SQL: :

DROP VIEW IF EXISTS us_origin_airport_SFO_global_tmp_view
DROP VIEW IF EXISTS us_origin_airport_JFK_tmp_view

Using SparkSession to Peruse Catalog

As mentioned above, Spark manages the metadata associated with each managed or unmanaged table. This is captured in the Catalog—a high-level abstraction in Spark SQL for storage of all metadata. This Catalog is accessible via programmatic interfaces.^{11 12}

For example, within a Spark application, after creating the SparkSession variable spark, you can access all metadata, such as databases, tables, columns, registered UDFs (user defined functions), locations, schema etc., through many of its public methods.

spark.catalog.listDatabases()
spark.catalog.listTables()
spark.catalog.listColumns(“us_delay_flights_tbl”)

Try importing the notebook from Github repo Learning Spark 2ed ¹³.

The spark.sql.Catalog API offers you the ability to peruse all metadata associated with your database and respective tables or views within the database. Not only can you create databases, tables, and views but can also inquire whether a table is cached or not.

Caching SQL Tables

Although we will discuss table caching strategies in the next chapter, it’s worth mentioning that like DataFrames, you can cache or uncache SQL tables and views too.

In SQL:

CACHE TABLE <table-name>
UNCACHE TABLE <table-name>

Reading Tables into DataFrames

Often, data engineers build data pipelines as part of their regular data ingestion and extract, transform, and load (ETL) process. They populate Spark SQL databases and tables with cleansed data for consumption by applications downstream.

Let’s assume an existing database learn_spark_db and a table us_delay_flights_tbl is ready for use. Instead of you reading from an external JSON file, you can simply use SQL to query the table and assign the returned result to a DataFrame.

// In Scala:
spark.sql(“USE learn_spark_db”)
val usFlightsDF =  spark.sql("SELECT * FROM us_delay_flights_tbl")
val usFlightsDF2 = spark.table(“us_delay_flights_tbl”)

# In Python
spark.sql(“USE learn_spark_db”)
us_flights_df = spark.sql("SELECT * FROM us_delay_flights_tbl")
us_flights_df2 = spark.table(“us_delay_flights_tbl”)

Now you have a cleansed DataFrame read from an existing Spark SQL database and table, created by a data engineer in your team. Furthermore, you can also read data from other file formats using the Spark built-in data sources, giving you the flexibility to interact with various common file formats.

DataFrameReader

DataFrameReader is the core construct for reading data from a data source into a DataFrame. It has a defined format and a recommended pattern for usage; this pattern of stringing methods together is a common usage in Spark. It also makes it easier to read. We have seen this pattern in Chapter 3 while exploring common data analysis patterns.

DataFrameReader.format(args).option(“key”, “value”).schema(args).load()

You can only access DataFrameReader through a SparkSession instance^{16 17}. That is, you cannot create an instance of DataFrameReader. To get an instance handle to it use:

SparkSession.read or SparkSession.readStream

While read returns a handle to DataStreamReader to read into a DataFrame from a static data source, readStream returns an instance to read from a streaming source. (We will cover Structured Streaming later in the book.)

Arguments to each of the methods to DataFrameReader take different values, and we enumerate each in the table below with a subset of the supported arguments.

format()

option()

schema()

DDL String or StructType, e.g., ‘A INT, B STRING’ or StructType(...)

load()

While we don’t expect to comprehensively enumerate all different combinations of arguments and options, the documentation for Python, Scala, R, and Java offer suggestions and guidance¹⁸. But it’s worthwhile to show a couple of examples:

// In Scala
// Use Parquet 
val file  = “/databricks-datasets/learning-spark-v2/data/flight-data/parquet/2010-summary.parquet”
 val df = spark.read.format(“parquet”)
.option(“path”, file)
.load() 
// Use the default format() value in the configuration
 val df2 = spark.read.option(“path”, file).load()
// Use CSV
val df3 = spark.read.format(“csv”)
    .option(“inferSchema”, “true”)
    .option(“header”, “true”)
    .option(“mode”, “PERMISSIVE”)
    .load(“/databricks-datasets/learning-spark-v2/data/flight-data/csv/*”)
// Use JSON
val df4 = spark.read.format(“json”)
        .option(“path”, “/databricks-datasets/learning-spark-v2/data/flight-data/json/*”)
        .load()

DataFrameWriter

Like its counterpart, DataFrameWriter does the reverse: it saves or writes data to a specified built-in data source. Unlike DataFrameReader, you access its instance not from a SparkSession but from the DataFrame you wish to save.^{19 20}

DataFrameWriter.format(args).option(args).bucketBy(args).partitionBy(args).save()
DataFrameWriter.format(args).option(args).sortBy(args).saveAsTable()

DataFrame.write or DataFrame.writeStream

Arguments to each of the methods to DataFrameWriter also take different values, and we enumerate each in the table below with a subset of the supported arguments:

format()

option()

bucketBy()

save() saveAsTableI()

A couple of short examples to illustrate the use of methods and arguments:

// In scala
// Use JSON
val location = …. 
df.write
  .format(“json”)
  .mode(“overwrite”)
  .sortBy(“lines”)
  .save(location)

Parquet

Let’s start with Parquet since it’s the default data source in Spark. Supported and widely used by many big data processing frameworks and platforms, Parquet is an open-source, columnar file format that offers many I/O optimizations such as compression and columnar format, which saves storage space and allows for quick access to data columns. ²¹

Because of its efficiency and I/O optimizations, we recommend that after you have transformed and cleansed your data, save your DataFrames in the Parquet format for downstream consumption. (Parquet is also the default table open format for Delta Lake, which we will cover in a later chapter.)

// In Scala
val file = “/databricks-datasets/learing-spark-v2/data/flight-data/parquet/2010-summary.parquet”
val df = spark.read.format(“parquet”).load(file)

# In Python
file = “...”
df = spark.read.format(“parquet”).load(file)

// In SQL
CREATE OR REPLACE TEMPORARY VIEW us_delay_flights_tbl
    USING parquet
    OPTIONS (
      path "/databricks-datasets/learing-spark-v2/data/flight-data/parquet/2010-summary.parquet" )

// In Scala/Python
spark.sql(“SELECT * FROM us_delay_flights_tbl”).show()
# In Python
spark.sql(“SELECT * FROM us_delay_flights_tbl”).show()

+-----------------+-------------------+-----+
|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|
+-----------------+-------------------+-----+
|United States    |Romania            |1    |
|United States    |Ireland            |264  |
|United States    |India              |69   |
|Egypt            |United States      |24   |
|Equatorial Guinea|United States      |1    |
|United States    |Singapore          |25   |
|United States    |Grenada            |54   |
|Costa Rica       |United States      |477  |
|Senegal          |United States      |29   |
|United States    |Marshall Islands   |44   |
+-----------------+-------------------+-----+
only showing top 10 rows

-rw-r--r--  1 jules  wheel    0 May 19 10:58 _SUCCESS
-rw-r--r--  1 jules  wheel  966 May 19 10:58 part-00000-bf73a6d3-ee4d-4087-ba4d-131e73ffddd4-c000.snappy.parquet

// In Scala/Python
df.write
  .mode(“overwrite”)
  .saveAsTable(“us_delay_flights_tbl”)

JSON

JavaScript Object Notation (JSON) is also a popular data format. It came to prominence as an easy-to-read-and-parse format compared to XML. It has two representational formats: single-line delimited mode and multi-line format mode. ²³ Both modes are supported in Spark.

In single line-delimited mode, each line denotes a single JSON object, whereas in a multi-line delimited mode, the entire multi-line object constitutes a single JSON object. To read in this mode set multiLine to true in the option() method.

val file = “/databricks-datasets/learning-spark-v2/data/flight-data/json/*
val df = spark
    .read
    .format(“json”)
    .option(“path”, file)
    .load()

# In Python
file = “/databricks-datasets/learning-spark-v2/data/flight-data/json/*”
df  = (spark
    .read
    .format(“json”)
    .option(“path”, file)
.load())

// In SQL

CREATE OR REPLACE TEMPORARY VIEW us_delay_flights_tbl
    USING json
    OPTIONS (
      path  “/databricks-datasets/learning-spark-v2/data/flight-data/json/*”
    )

// In Scala/Python
spark.sql(“SELECT * FROM us_delay_flights_tbl”).show()


+-----------------+-------------------+-----+
|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|
+-----------------+-------------------+-----+
|United States    |Romania            |15   |
|United States    |Croatia            |1    |
|United States    |Ireland            |344  |
|Egypt            |United States      |15   |
|United States    |India              |62   |
|United States    |Singapore          |1    |
|United States    |Grenada            |62   |
|Costa Rica       |United States      |588  |
|Senegal          |United States      |40   |
|Moldova          |United States      |1    |
+-----------------+-------------------+-----+
only showing top 10 rows

// In Scala
df.write.format(“json”)
.mode(“overwrite”)
.options(“path”, “/tmp/data/json/df_json”)
.option(“compression”, “snappy”)
.save()

-rw-r--r--  1 jules  wheel   0 May 16 14:44 _SUCCESS
-rw-r--r--  1 jules  wheel  71 May 16 14:44 part-00000-5149e5b3-fe04-410d-bca6-f92a3cbc81ec-c000.json

CSV

As widely used as a plain text file, this common text file format captures each datum or field delimited by a comma; each line with comma separated fields represents a record. Even though a comma is the default separator, you may use other delimiters to separate fields, in cases where a comma is part of your data. For example, a series of temperatures for a day could be separated by a comma. Popular spreadsheets can generate CSV files, thus its common usage among data and business analysts.

// In Scala
val file = “/databricks-datasets/learning-spark-v2/data/flight-data/csv/*”
val schema = "DEST_COUNTRY_NAME STRING, ORIGIN_COUNTRY_NAME STRING, count INT"

val df = spark
    .read
    .format(“csv”)
    .schema(schema)
    .option(“header”, “true”)
    .option(“mode”, “FAILFAST”)     // exit if any errors
    .option(“nullValue”, “”)        // replace any null data field with “”
    .option(“path”, file)
    .load()

# In Python
file = “/databricks-datasets/learning-spark-v2/data/flight-data/csv/*”
schema = "DEST_COUNTRY_NAME STRING, ORIGIN_COUNTRY_NAME STRING, count INT"
df  = (spark
    .read
    .format("csv")
    .option("header", "true")
    .schema(schema)
    .option("mode", "FAILFAST")      # exit if any errors
    .option("nullValue", "")      # replace any null data field with “”
    .option("path", file)
    .load())

// In SQL
CREATE or REPLACE TEMPORARY VIEW us_delay_flights_tbl
    USING csv
    OPTIONS (
      path "/databricks-datasets/LearingSpark-v2/data/flight-data/csv/*",
    header “true”,
      inferSchema “true”,
    mode “FAILFAST”
    )

// In Scala
spark.sql(“SELECT * FROM us_delay_flights_tbl”).show()
# In Python
df = spark.sql(“SELECT * FROM us_delay_flights_tbl”).show()
+-----------------+-------------------+-----+
|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|
+-----------------+-------------------+-----+
|United States    |Romania            |1    |
|United States    |Ireland            |264  |
|United States    |India              |69   |
|Egypt            |United States      |24   |
|Equatorial Guinea|United States      |1    |
|United States    |Singapore          |25   |
|United States    |Grenada            |54   |
|Costa Rica       |United States      |477  |
|Senegal          |United States      |29   |
|United States    |Marshall Islands   |44   |
+-----------------+-------------------+-----+
only showing top 10 rows

// In Scala

df.write.format(“csv”)
.mode(“overwrite”)
.options(“path”, “/tmp/data/csv/df_csv”)
.save()

-rw-r--r--  1 jules  wheel   0 May 16 12:17 _SUCCESS
-rw-r--r--  1 jules  wheel  36 May 16 12:17 part-00000-251690eb-682c-4a6d-8c42-d6abfa663125-c000.csv

Avro

Introduced in Spark 2.4 as a built-in data source, Avro format is used by Apache Kafka for message serializing and deserializing for many reasons: direct mapping to JSON; efficient and fast; and bindings available for many programming languages.^{30 31}

// In Scala
val df = spark.read
.format(“avro”)
.option(“path”, “/tmp/data/avro/episodes.avro”)
.load()

df.show(false)

# In Python
df = (spark.read
.format(“avro”)
.option(“path”, “/tmp/data/avro/episodes.avro”)
.load())

 df.show(False)
+-----------------------+----------------+------+
|title                  |air_date        |doctor|
+-----------------------+----------------+------+
|The Eleventh Hour      |3 April 2010    |11    |
|The Doctor's Wife      |14 May 2011     |11    |
|Horror of Fang Rock    |3 September 1977|4     |
|An Unearthly Child     |23 November 1963|1     |
|The Mysterious Planet  |6 September 1986|6     |
|Rose                   |26 March 2005   |9     |
|The Power of the Daleks|5 November 1966 |2     |
|Castrolava             |4 January 1982  |5     |
+-----------------------+----------------+------+

CREATE or REPLACE TEMPORARY VIEW episode_tbl
    USING avro
    OPTIONS (
      path "/tmp/data/avro/episodes.avro”
    )

// In Scala
spark.sql(“SELECT * FROM episode_tbl”).show(false)
// In Python
spark.sql(“SELECT * FROM episode_tbl”).show(False)

+-----------------------+----------------+------+
|title                  |air_date        |doctor|
+-----------------------+----------------+------+
|The Eleventh Hour      |3 April 2010    |11    |
|The Doctor's Wife      |14 May 2011     |11    |
|Horror of Fang Rock    |3 September 1977|4     |
|An Unearthly Child     |23 November 1963|1     |
|The Mysterious Planet  |6 September 1986|6     |
|Rose                   |26 March 2005   |9     |
|The Power of the Daleks|5 November 1966 |2     |
|Castrolava             |4 January 1982  |5     |
+-----------------------+----------------+------+

// In Scala
df.write
.format("avro")
.mode("overwrite")
.option("path", "/tmp/data/episodes_avro")
.save()

-rw-r--r--  1 jules  wheel    0 May 17 11:54 _SUCCESS
-rw-r--r--  1 jules  wheel  526 May 17 11:54 part-00000-ffdf70f4-025f-4ec1-b25e-4e2fbade0312-c000.avro

ORC

As an additional optimized columnar file format, Spark 2.x supports vectorized ORC reader. Two Spark configurations dictate which ORC implementation to use. When Spark configurations spark.sql.orc.impl is set to native and spark.sql.orc.enableVectorizedReader is set to true, Spark uses the vectorized ORC reader. A vectorized reader reads blocks of rows (often 1024 per block) instead of one row at a time, streamlining operations and reducing CPU usage for intensive operations like scans, filters, aggregations, and joins.³⁴

For creating Hive ORC serde tables, especially in the SQL command USING HIVE options, vectorized reader is used when the Spark configuration spark.sql.hive.convertMetastoreOrc is set to true. Serde is serialization and deserialization FileFormat used by Hive tables. ^{35 36}

// In Scala: 
val file = “/databricks-datasets/learing-spark-v2/data/flight-data/orc/*.*
val df = spark.read
.format(“orc”)
.option(“path”, file)
.load()

df.show(10, false)

# In Python
file  = “/databricks-datasets/learning-spark-v2/data/flight-data/orc/*.*”
df = (spark.read
.format(“orc”)
.option(“path”, file)
.load())

 df.show(10, False)

+-----------------+-------------------+-----+
|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|
+-----------------+-------------------+-----+
|United States    |Romania            |1    |
|United States    |Ireland            |264  |
|United States    |India              |69   |
|Egypt            |United States      |24   |
|Equatorial Guinea|United States      |1    |
|United States    |Singapore          |25   |
|United States    |Grenada            |54   |
|Costa Rica       |United States      |477  |
|Senegal          |United States      |29   |
|United States    |Marshall Islands   |44   |
+-----------------+-------------------+-----+
only showing top 10 rows

// In SQL
CREATE or REPLACE TEMPORARY VIEW us_delay_flights_tbl
    USING orc
    OPTIONS (
      path “/databricks-datasets/learning-spark-v2/data/flight-data/orc/*.*”
    )

// In Scala/Python
spark.sql(“SELECT * FROM us_delay_flights_tbl”).show()

+-----------------+-------------------+-----+
|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|
+-----------------+-------------------+-----+
|United States    |Romania            |1    |
|United States    |Ireland            |264  |
|United States    |India              |69   |
|Egypt            |United States      |24   |
|Equatorial Guinea|United States      |1    |
|United States    |Singapore          |25   |
|United States    |Grenada            |54   |
|Costa Rica       |United States      |477  |
|Senegal          |United States      |29   |
|United States    |Marshall Islands   |44   |
+-----------------+-------------------+-----+
only showing top 10 rows

// In Scala
df.write.format(“orc”)
.mode(“overwrite”)
.option(“path”, “/tmp/data/orc/df_orc”)
.option(“compression”, “snappy”)
.save()

-rw-r--r--  1 jules  wheel    0 May 16 17:23 _SUCCESS
-rw-r--r--  1 jules  wheel  547 May 16 17:23 part-00000-bd41c3ec-38d8-4263-af46-6a0b3150c5ab-c000.snappy.orc

Image

In Spark 2.x, the community introduced a new Data Source, image files, to support deep learning and machine learning frameworks such as TensorFlow, PyTorch, and Deep Learning Pipelines as first-class citizens of Apache Spark.^{37 38} For computer vision machine learning applications, loading and processing image datasets is important.

In Scala:

import org.apache.spark.ml.source.image

val imageDir = “/databricks-datasets/cctvVideos/train_images/”
val imagesDF = spark.read.format(“image”).load(image_dir)

imagesDF.printSchema
imagesDF.select("image.height", "image.width", "image.nChannels", "image.mode", "label").show(5, false)

In Python:

from pyspark.ml.source import image

image_dir = “/databricks-datasets/cctvVideos/train_images/”
images_df = spark.read.format(“image”).load(image_dir)
images_df.printSchema()

images_df.select("image.height", "image.width", "image.nChannels", "image.mode", "label").show(5, truncate=False)

root
 |-- image: struct (nullable = true)
 |    |-- origin: string (nullable = true)
 |    |-- height: integer (nullable = true)
 |    |-- width: integer (nullable = true)
 |    |-- nChannels: integer (nullable = true)
 |    |-- mode: integer (nullable = true)
 |    |-- data: binary (nullable = true)
 |-- label: integer (nullable = true)


+------+-----+---------+----+-----+
|height|width|nChannels|mode|label|
+------+-----+---------+----+-----+
|288   |384  |3        |16  |0    |
|288   |384  |3        |16  |1    |
|288   |384  |3        |16  |0    |
|288   |384  |3        |16  |0    |
|288   |384  |3        |16  |0    |
+------+-----+---------+----+-----+
only showing top 5 rows