To load text files, we use the text method, which will return a single column with the column name set to value and type as string. Note that the Dataset API is not supported in Python language. So, in Python language, a DataFrame is returned while in Scala language Dataset it is returned, as shown in the following:

>>> df_txt = spark.read.text("people.txt")
>>> df_txt.show()
+-----------+
|      value|
+-----------+
|Michael, 29|
|   Andy, 30|
| Justin, 19|
+-----------+
>>> df_txt
DataFrame[value: string]

Spark SQL can automatically infer the schema of a JSON dataset when loading to a DataFrame:

>>> df_json = spark.read.json("people.json")
>>> df_json.printSchema()
root
 |-- age: long (nullable = true)
 |-- name: string (nullable = true)

You can also use the load method to load files. The following command will throw an exception with the message NOT A PARQUET FILE. Because, the default data source is configured to parquet. So, to read a .json file, use format to specify the type of datasource:

>>> df_json = spark.read.load("people.json")

Manually specify the type of file with format:

>>> df_json = spark.read.load("people.json", format="json")

To write data to another JSON file, use one of the following commands:

>>> df_json.write.json("newjson_dir")
>>> df_json.write.format("json").save("newjson_dir2")

To write data to any other format, just mention the format you want to save. The following example saves the df_json DataFrame in Parquet format:

>>> df_json.write.parquet("parquet_dir")
>>> df_json.write.format("parquet").save("parquet_dir2")

Apache Parquet is a columnar storage format that provides superior performance and is available in any Hadoop project. File format documentation can be found at https://parquet.apache.org/documentation/latest/.

As mentioned earlier, whatever the type of data source, similar read and write functions of DataSources API are used. So, let's use read and write functions for parquet files in the parquet_dir directory as well.

To create a DataFrame from a Parquet file and issue SQL commands, use the following command:

>>> df_parquet = spark.read.load("parquet_dir")

Or use the following command:

>>> df_parquet = spark.read.parquet("parquet_dir")

Note that format is not needed when using the load method because parquet is the default data source. This reads all parquet files from the parquet_dir directory:

>>> df_parquet.createOrReplaceTempView("parquet_table");
>>> teenagers = spark.sql("SELECT name from parquet_table where age >= 13 AND age <= 19")

To write the data from a parquet DataFrame to .json format, use one of the write commands:

>>> df_parquet.write.json("myjson_dir")
>>> df_parquet.write.format("json").save("myjson_dir2")

There are multiple modes while writing data. By default, error mode is enabled, which will throw an error if output already exists. Other modes are append, overwrite, and ignore when the target data source already exists. The following example appends data to the myjson_dir and parquet_dir directory:

>>> df_parquet.write.mode("append").json("myjson_dir")
>>> df_parquet.write.mode("append").save("parquet_dir")

A DataFrame can be written to a Hive Table. The following example writes data to a hive managed table in the default Parquet format:

>>> df_parquet.write.saveAsTable("hive_parquet_table")

Parquet supports schema evolution such as ProtocolBuffer, Avro, and Thrift. Multiple schemas can be merged by setting the global option, spark.sql.parquet.mergeSchema to true or using the mergeSchema option as shown in the following:

>>> df_parquet = spark.read.option("mergeSchema", "true").parquet("parquet_dir")

Table partitioning is a common optimization approach used in parquet hive tables as well. Partition discovery and partition pruning happen automatically for queries performed on Parquet tables. Also, while writing the parquet tables, data can be partitioned using the partitionBy method.

Since Parquet is a columnar format, column pruning is automatic and also predicate pushdown is automatic by default with the spark.sql.parquet.filterPushdown parameter set to true.

The Optimized Row Columnar (ORC) file format provides an efficient way to store data. More details on file format and documentation can be found at https://cwiki.apache.org/confluence/display/Hive/LanguageManual+ORC.

Column Pruning: Since ORC is a columnar format, columns needed by the query are read-only.

Predicate Push-down: ORC enables predicate pushdown with the following parameter (with the help of indexes within each file). Check the current configuration and then change the value:

>>> spark.conf.get("spark.sql.orc.filterPushdown")
u'false'
>>> spark.conf.set("spark.sql.orc.filterPushdown","true")
>>> spark.conf.get("spark.sql.orc.filterPushdown")
u'true'

Partition Pruning: Just like Parquet tables, partition discovery and partition pruning are automatic for ORC tables too. To create ORC files, partitioning data by a column, use the following syntax:

>>> df_json.write.format("orc").partitionBy("age").save("partitioned_orc")

Spark SQL provides easy integration with external databases using JDBC. To use this functionality, you will need to include the JDBC driver for your particular database on the spark classpath. For example, in order to connect to a Postgres database from the Spark shell, you will use the following command or copy the jdbc jar to the jars directory of SPARK_HOME:

[cloudera@quickstart spark-2.0.0-bin-hadoop2.7 ]$  SPARK_CLASSPATH=postgresql-9.3-1102-jdbc41.jar bin/pyspark

Tables from external databases can be loaded into Spark SQL as a DataFrame or as a temporary table. The following options are supported while creating a DataFrame using jdbc:

Let's create a DataFrame using one of the two commands given in the following:

>>> df = spark.read.format('jdbc').options(url='jdbc:mysql://localhost:3306/retail_db?user=root&password=cloudera', dbtable='departments').load()
>>> df = spark.read.format('jdbc').options(url='jdbc:mysql://localhost:3306/retail_db', dbtable='departments', user='root', password='cloudera').load()
>>> df.show()

Then, register it as a temporary table and query it:

>>> df.createTempView("dept")
>>> df_new = spark.sql("select * from dept where department_id> 5")

Let's write the DataFrame to another table in the MySQL database. This will create the table and write the data out:

>>> df_new.write.jdbc("jdbc:mysql://localhost:3306/retail_db?user=root&password=cloudera","new_table")

To write the DataFrame to another table in the MySQL database (and also specify the overwrite option to overwrite the data if the table is already existing):

>>> df_new.write.jdbc("jdbc:mysql://localhost:3306/retail_db?user=root&password=cloudera","new_table","overwrite")

Download a .csv dataset as shown in the following and then copy it to HDFS. Start a pyspark shell to create a DataFrame from a .csv file:

[cloudera@quickstart spark-2.0.0-bin-hadoop2.7 ]$  wget https://raw.githubusercontent.com/databricks/spark-csv/master/src/test/resources/cars.csv --no-check-certificate

[cloudera@quickstart spark-2.0.0-bin-hadoop2.7 ]$ hadoop fs -put cars.csv

>>> csv_df = spark.read.options(header='true',inferSchema='true').csv("cars.csv")
>>> csv_df.printSchema()
root
 |-- year: integer (nullable = true)
 |-- make: string (nullable = true)
 |-- model: string (nullable = true)
 |-- comment: string (nullable = true)
 |-- blank: string (nullable = true)

Let's select a few columns and write the DataFrame to another CSV file with a different compression codec.

>>> csv_df.select('year', 'model').write.options(codec="org.apache.hadoop.io.compress.GzipCodec").csv('newcars.csv')

Let's start the pyspark shell by providing the --packages dependency:

[cloudera@quickstart spark-2.0.0-bin-hadoop2.7 ]~  bin/pyspark --packages com.databricks:spark-avro_2.11:3.0.0

Let's read the JSON file, write it in AVRO format, and then read the same file:

>>> df_json = spark.read.json("people.json")
>>> df_json.write.format("com.databricks.spark.avro").save("avro_out")
>>> df_avro = spark.read.format("com.databricks.spark.avro").load("avro_out")

You can specify the record name and namespace to use when writing out by passing recordName and recordNamespace as optional parameters to Spark:

>>> df_avro.write.format("com.databricks.spark.avro").option("recordName","AvroTest").option("recordNamespace","com.cloudera.spark").save("newavro")

Reading XML data will be same as AVRO where we include external .xml package with pyspark shell. Perform the following steps to download .xml file and work with it:

Pandas is a Python library for data manipulation. Use the following commands to create a DataFrame in Pandas and convert it to Spark DataFrame and vice versa. Install pandas using pip if it is not done already:

>>> import pandas
>>> data = [
('v1', 'v5', 'v9'),
('v2', 'v6', 'v10'),
('v3', 'v7', 'v11'),
('v4', 'v8', 'v12')]

>>> pandas_df = pandas.DataFrame(data, columns=['col1', 'col2', 'col3'])

>>> spark_df = spark.createDataFrame(pandas_df)

>>> spark_df.toPandas()

HBase does not offer SQL as a means to query the data. It exposes GET, SCAN, and FILTER APIs, which are difficult to implement in ad-hoc queries. So usually, for analyzing the data on HBase, it is common to write MapReduce programs using Java. Integration with Hive or Pig provides an easy way to write SQLs or scripts to analyze the data, but they are not efficient since they use TableInputFormat instead of the native HBase API. Apache Phoenix provides an easy way to integrate with HBase and write relational queries. Spark-on-HBase connector is a new connector based on the DataFrame framework and is created using the standard Datasource API so it can be simply used by specifying the packages option. It does not leverage TableInputFormat, but it creates a customized RDD that can implement partition and column pruning, predicate push down, and data locality. It also leverages the Spark Catalyst optimizer for optimizations. The following exercise was tried on HDP 2.5 sandbox version with Spark version 1.6.2.

Let's get started with Spark-on-HBase connector using the following steps: