Exploring Stock Prices Using Spark SQL¶
Introduction¶
In this project, distributed programming using Spark will be implemented and then derive knowledge from data. The data will be loaded for analysis, and will be explored by using Spark RDD, Spark DataFrames. Explored more and performed statistical analysis on Stock prices datasets using Apache Spark SQL and Spark DataFrame API. Lastly, a parquet tables will be created and then store their results in them.
Apache Spark
- Open source distributed cluster computing framework.
- Provides an interface for programming entire clusters with implicit parallelism and fault tolerance
Use cases:
- Interactive data analysis at scale
- Streaming data analysis
- Machine learning pipelines
- ETL (Extract-Transform-Load)
Dataset:
- Tesla Stock Prices
- Amazon Stock Prices
- Google Stock Prices
Source:
- Yahoo Finance
Spark Application And Spark Context¶
In [3]:
# Import library
from pyspark import SparkContext
# Create shorcut term
sc = SparkContext.getOrCreate()
In [4]:
# Check type
type(sc)
Out[4]:
Load data into spark¶
In [5]:
# Define collections of numbers
data = [1, 2, 3, 4, 5]
# Parallel RDD
distData = sc.parallelize(data)
In [6]:
# Import library
from pyspark.sql import SQLContext
# label SQL Context
sqlContext = SQLContext(sc)
In [7]:
# Create list
l = [('Alice', 1)]
# Create dataframe
df = sqlContext.createDataFrame(l, ['name', 'age'])
# Show data
df.collect()
Out[7]:
In [8]:
# Label data
data_file = 'TSLA.csv'
# Read the csv
raw_rdd = sc.textFile(data_file).cache()
# Return the five element in the data
raw_rdd.take(5)
Out[8]:
In [9]:
# Check type
type(raw_rdd)
Out[9]:
In [10]:
# Apply map transformation
csv_rdd = raw_rdd.map(lambda row: row.split(","))
# Show data
print(csv_rdd.take(2))
# Check data type
print(type(csv_rdd))
In [11]:
# Check length
len(csv_rdd.take(1)[0])
Out[11]:
Spark DataFrames¶
- Immutable distributed collections of data
- Data organized into named columns
Operations:
- Filter
- Group By
- Compute Aggregation
- SQL Queries
In [12]:
# Create data frame
df = sqlContext.read.load(data_file,
format='com.databricks.spark.csv',
header='true',
inferSchema='true')
In [13]:
# Check count
df.count()
Out[13]:
In [14]:
# Check data
df.take(5)
Out[14]:
In [15]:
# Check schema
df.printSchema()
In [16]:
# Import library
import pandas
# Check pandas dataframe
df.toPandas().head(5)
Out[16]:
Explore And Query Data¶
In [17]:
# Import library
from pyspark.sql.types import *
# Set up structure type
schema = StructType([
StructField("date", StringType(), True),
StructField("openprice", IntegerType(), True),
StructField("highprice", IntegerType(), True),
StructField("lowprice", IntegerType(), True),
StructField("closeprice", IntegerType(), True),
StructField("volume", IntegerType(), True),
StructField("adjcloseprice", IntegerType(), True)])
# Create dataframe
df2 = sqlContext.read.load(data_file,
format='com.databricks.spark.csv',
header='true',
schema=schema)
In [18]:
# Check dataframe
df2.take(2)
Out[18]:
In [19]:
# Check schema
df2.printSchema()
In [20]:
# Check type
type(df)
Out[20]:
In [21]:
# Show data
df.show(5)
In [22]:
# Label data
tesla_data_file = 'TSLA.csv'
# Create dataframe
tesla_df = sqlContext.read.load(tesla_data_file,
format='com.databricks.spark.csv',
header='true',
inferSchema='true')
# Show data
tesla_df.show()
In [23]:
# Label data
google_data_file = 'GOOG.csv'
# Create dataframe
google_df = sqlContext.read.load(google_data_file,
format='com.databricks.spark.csv',
header='true',
inferSchema='true')
# Show data
google_df.show()
In [24]:
# Label data
amazon_data_file = 'AMZN.csv'
# Create dataframe
amazon_df = sqlContext.read.load(amazon_data_file,
format='com.databricks.spark.csv',
header='true',
inferSchema='true')
# Show data
amazon_df.show()
Spark SQL¶
In [26]:
# Import libraries
from pyspark.sql.functions import year, month, dayofmonth
import datetime
# Average closing price per year for AMZN
# stocksDF.select(year($"dt").alias("yr"), $"adjcloseprice").groupBy("yr").avg("adjcloseprice").orderBy(desc("yr")).show
amazon_df.select(year("Date").alias("year"), "AdjClose").groupby("year").avg("AdjClose").sort("year").show()
In [27]:
# Compute the average closing price per month for apc
# stocksDF.select(year($"dt").alias("yr"),month($"dt").alias("mo"), $"adjcloseprice")
# .groupBy("yr","mo").agg(avg("adjcloseprice")).orderBy(desc("yr"),desc("mo")).show
amazon_df.select(year("Date").alias("year"),
month("Date").alias("month"),
"AdjClose").groupby("year", "month").avg("AdjClose").sort("year", "month").show()
In [29]:
# Register the DataFrames as temp views
amazon_df.registerTempTable("amazon_stocks")
google_df.registerTempTable("google_stocks")
tesla_df.registerTempTable("tesla_stocks")
In [30]:
# Query entry
sqlContext.sql("SELECT * FROM amazon_stocks").show(5)
In [31]:
# Calculate and display the average closing price per month for XOM ordered by year,month
sqlContext.sql("""SELECT year(amazon_stocks.Date) as yr, month(amazon_stocks.Date) as mo, avg(amazon_stocks.AdjClose) from amazon_stocks group By year(amazon_stocks.Date), month(amazon_stocks.Date)""").show()
In [32]:
# When did the closing price for SPY go up or down by more than 2 dollars?
sqlContext.sql("SELECT google_stocks.Date, google_stocks.Open, google_stocks.Close, abs(google_stocks.Close - google_stocks.Open) as spydif FROM google_stocks WHERE abs(google_stocks.Close - google_stocks.Open) > 4 ").show()
In [33]:
# What was the max, min closing price for SPY and XOM by Year?</font>
sqlContext.sql("SELECT year(tesla_stocks.Date) as yr, max(tesla_stocks.AdjClose), min(tesla_stocks.AdjClose) FROM tesla_stocks group By year(tesla_stocks.Date)").show()
In [34]:
# Check physical plan
sqlContext.sql("SELECT year(tesla_stocks.Date) as yr, max(tesla_stocks.AdjClose), min(tesla_stocks.AdjClose) FROM tesla_stocks group By year(tesla_stocks.Date)").explain()
In [35]:
# Join all stock closing prices in order to compare
joinclose=sqlContext.sql("SELECT tesla_stocks.Date, tesla_stocks.AdjClose as teslaclose, amazon_stocks.AdjClose as amazonclose, google_stocks.AdjClose as googleclose from tesla_stocks join google_stocks on tesla_stocks.Date = google_stocks.Date join amazon_stocks on tesla_stocks.Date = amazon_stocks.Date").cache()
# Show data frame
joinclose.show()
# Create temporary table
joinclose.registerTempTable("joinclose")
In [36]:
# Check average close price by year
sqlContext.sql("SELECT year(joinclose.Date) as yr, avg(joinclose.teslaclose) as teslaclose, avg(joinclose.amazonclose) as amazonclose, avg(joinclose.googleclose) as googleclose from joinclose group By year(joinclose.Date) order by year(joinclose.Date)").show()
Save Spark DataFrames¶
In [37]:
# Save as parquet file
joinclose.write.format("parquet").save("joinstocks.parquet")
In [38]:
# Import parquet file beck to data frame
final_df = sqlContext.read.parquet("joinstocks.parquet")
# Show data frame
final_df.show()
In [39]:
# Check schema
final_df.printSchema()
In [40]:
# Average of each by month
final_df.select(year("Date").alias("year"),
month("Date").alias("month"),
"teslaclose", "amazonclose",
"googleclose").groupby("year",
"month").avg("teslaclose",
"amazonclose",
"googleclose").sort("year",
"month").show()
In [41]:
# Check physical plan
final_df.select(year("Date").alias("year"),
month("Date").alias("month"),
"teslaclose", "amazonclose",
"googleclose").groupby("year",
"month").avg("teslaclose",
"amazonclose",
"googleclose").sort("year",
"month").explain()