Prepare Datasets for JoinsΒΆ
Let us prepare datasets to join. We will be using Airtraffic data as well as retail data while going through the tasks.
Make sure airport-codes is in HDFS.
We will also use airtraffic data for the month of January 2008. We have used that data set in the past as well.
We will be using retail data using JSON format. Make sure you also have the retail data using JSON format in appropriate location.
Let us start spark context for this Notebook so that we can execute the code provided. You can sign up for our 10 node state of the art cluster/labs to learn Spark SQL using our unique integrated LMS.
from pyspark.sql import SparkSession
import getpass
username = getpass.getuser()
spark = SparkSession. \
builder. \
config('spark.ui.port', '0'). \
config("spark.sql.warehouse.dir", f"/user/{username}/warehouse"). \
enableHiveSupport(). \
appName(f'{username} | Python - Joining Data Sets'). \
master('yarn'). \
getOrCreate()
If you are going to use CLIs, you can use Spark SQL using one of the 3 approaches.
Using Spark SQL
spark2-sql \
--master yarn \
--conf spark.ui.port=0 \
--conf spark.sql.warehouse.dir=/user/${USER}/warehouse
Using Scala
spark2-shell \
--master yarn \
--conf spark.ui.port=0 \
--conf spark.sql.warehouse.dir=/user/${USER}/warehouse
Using Pyspark
pyspark2 \
--master yarn \
--conf spark.ui.port=0 \
--conf spark.sql.warehouse.dir=/user/${USER}/warehouse
!hdfs dfs -ls /public/airtraffic_all
Found 3 items
drwxr-xr-x - hdfs supergroup 0 2021-03-02 19:48 /public/airtraffic_all/airport-codes
drwxr-xr-x - hdfs supergroup 0 2021-03-02 19:42 /public/airtraffic_all/airtraffic
drwxr-xr-x - hdfs supergroup 0 2021-03-02 19:48 /public/airtraffic_all/airtraffic-part
!hdfs dfs -ls /public/airtraffic_all/airport-codes
Found 1 items
-rw-r--r-- 3 hdfs supergroup 11411 2021-03-02 19:48 /public/airtraffic_all/airport-codes/airport-codes-na.txt
airportCodesPath = "/public/airtraffic_all/airport-codes"
airportCodes = spark. \
read. \
option("sep", "\t"). \
option("header", True). \
option("inferSchema", True). \
csv(airportCodesPath)
airportCodes.printSchema()
root
|-- City: string (nullable = true)
|-- State: string (nullable = true)
|-- Country: string (nullable = true)
|-- IATA: string (nullable = true)
airportCodes.show()
+-----------+-----+-------+----+
| City|State|Country|IATA|
+-----------+-----+-------+----+
| Abbotsford| BC| Canada| YXX|
| Aberdeen| SD| USA| ABR|
| Abilene| TX| USA| ABI|
| Akron| OH| USA| CAK|
| Alamosa| CO| USA| ALS|
| Albany| GA| USA| ABY|
| Albany| NY| USA| ALB|
|Albuquerque| NM| USA| ABQ|
| Alexandria| LA| USA| AEX|
| Allentown| PA| USA| ABE|
| Alliance| NE| USA| AIA|
| Alpena| MI| USA| APN|
| Altoona| PA| USA| AOO|
| Amarillo| TX| USA| AMA|
|Anahim Lake| BC| Canada| YAA|
| Anchorage| AK| USA| ANC|
| Appleton| WI| USA| ATW|
| Arviat| NWT| Canada| YEK|
| Asheville| NC| USA| AVL|
| Aspen| CO| USA| ASE|
+-----------+-----+-------+----+
only showing top 20 rows
airportCodes.count()
526
!hdfs dfs -ls /public/airtraffic_all/airtraffic-part/flightmonth=200801
Found 1 items
-rw-r--r-- 3 hdfs supergroup 14654075 2021-03-02 19:47 /public/airtraffic_all/airtraffic-part/flightmonth=200801/part-00252-5cde1303-4ebf-4a12-8fad-f5d9f9c9124a.c000.snappy.parquet
!hdfs dfs -find /public/retail_db_json
/public/retail_db_json
/public/retail_db_json/categories
/public/retail_db_json/categories/_SUCCESS
/public/retail_db_json/categories/part-r-00000-ce1d8208-178d-48d3-bfb2-1a97d9c05094
/public/retail_db_json/customers
/public/retail_db_json/customers/_SUCCESS
/public/retail_db_json/customers/part-r-00000-70554560-527b-44f6-9e80-4e2031af5994
/public/retail_db_json/departments
/public/retail_db_json/departments/_SUCCESS
/public/retail_db_json/departments/part-r-00000-3db7cfae-3ad2-4fc7-88ff-afe0ec709f49
/public/retail_db_json/order_details
/public/retail_db_json/order_details/order_details.json
/public/retail_db_json/order_items
/public/retail_db_json/order_items/_SUCCESS
/public/retail_db_json/order_items/part-r-00000-6b83977e-3f20-404b-9b5f-29376ab1419e
/public/retail_db_json/orders
/public/retail_db_json/orders/_SUCCESS
/public/retail_db_json/orders/part-r-00000-990f5773-9005-49ba-b670-631286032674
/public/retail_db_json/products
/public/retail_db_json/products/_SUCCESS
/public/retail_db_json/products/part-r-00000-158b7037-4a23-47e6-8cb3-8cbf878beff7
orders = spark.read.json('/public/retail_db_json/orders')
orders.printSchema()
root
|-- order_customer_id: long (nullable = true)
|-- order_date: string (nullable = true)
|-- order_id: long (nullable = true)
|-- order_status: string (nullable = true)
orders.show()
+-----------------+--------------------+--------+---------------+
|order_customer_id| order_date|order_id| order_status|
+-----------------+--------------------+--------+---------------+
| 11599|2013-07-25 00:00:...| 1| CLOSED|
| 256|2013-07-25 00:00:...| 2|PENDING_PAYMENT|
| 12111|2013-07-25 00:00:...| 3| COMPLETE|
| 8827|2013-07-25 00:00:...| 4| CLOSED|
| 11318|2013-07-25 00:00:...| 5| COMPLETE|
| 7130|2013-07-25 00:00:...| 6| COMPLETE|
| 4530|2013-07-25 00:00:...| 7| COMPLETE|
| 2911|2013-07-25 00:00:...| 8| PROCESSING|
| 5657|2013-07-25 00:00:...| 9|PENDING_PAYMENT|
| 5648|2013-07-25 00:00:...| 10|PENDING_PAYMENT|
| 918|2013-07-25 00:00:...| 11| PAYMENT_REVIEW|
| 1837|2013-07-25 00:00:...| 12| CLOSED|
| 9149|2013-07-25 00:00:...| 13|PENDING_PAYMENT|
| 9842|2013-07-25 00:00:...| 14| PROCESSING|
| 2568|2013-07-25 00:00:...| 15| COMPLETE|
| 7276|2013-07-25 00:00:...| 16|PENDING_PAYMENT|
| 2667|2013-07-25 00:00:...| 17| COMPLETE|
| 1205|2013-07-25 00:00:...| 18| CLOSED|
| 9488|2013-07-25 00:00:...| 19|PENDING_PAYMENT|
| 9198|2013-07-25 00:00:...| 20| PROCESSING|
+-----------------+--------------------+--------+---------------+
only showing top 20 rows
orders.count()
68883