{"id":261,"date":"2018-01-21T19:11:01","date_gmt":"2018-01-21T19:11:01","guid":{"rendered":"http:\/\/eipsoftware.com\/musings\/?p=261"},"modified":"2018-02-20T19:24:43","modified_gmt":"2018-02-20T19:24:43","slug":"spark-sqlcontext","status":"publish","type":"post","link":"https:\/\/eipsoftware.com\/musings\/spark-sqlcontext\/","title":{"rendered":"Spark – SQLContext"},"content":{"rendered":"

Spark – SQLContext<\/h4>\n

Today we will look at the SQLContext object from the PySpark library and how you can use it to connect to a local database.\u00a0 In the example below we will:
\nConnect to a local PostgreSQL database and read the contents into a dataframe.
\nRun some simple SQL queries
\nAnd join two data frames together<\/p>\n

Let’s get started.<\/p>\n

<\/p>\n

Initializing Environment<\/strong><\/h5>\n

In the environment I was using I had a local instance of a PostgreSQL database installed.<\/p>\n

I had a database called “Cloudera”. You can see all available databases in PostgreSQL by using the \\list<\/strong> command from a PostgreSQL prompt. There are 3 tables in the database, adclicks, buyclicks, and gameclicks.\u00a0 Use the \\dt<\/strong> command to show the tables in the database.<\/p>\n

Python Code<\/h5>\n

First we need to import the object from the PySpark library. And will create an object of type SQLContext named sqlsc.<\/p>\n

from pyspark.sql import SQLContext\r\nsqlsc = SQLContext(sc)<\/pre>\n
Next will read the contents of the gameclicks table and store the result into a dataframe named df. And then we can see the results of the schema. I am using jdbc:postgresql driver to connect to the database.<\/p>\n
df = sqlsc.read.format(\"jdbc\") \\\r\n  .option(\"url\", \"jdbc:postgresql:\/\/localhost\/cloudera?user=cloudera\") \\\r\n  .option(\"dbtable\", \"gameclicks\") \\\r\n  .load()\r\n\r\ndf.printSchema()<\/pre>\n
The gameclicks schema is below.<\/p>\n
\"\"<\/p>\n
 <\/p>\n
Now, we will count how many rows are in the data frame and show the top 5 rows.<\/p>\n
df.count()\r\n\r\ndf.show(5)<\/pre>\n
755806<\/pre>\n
 <\/p>\n
\"\"<\/p>\n
PySpark Queries<\/h5>\n
Now that the data frame is loaded we can perform some simple queries.<\/p>\n
First we will do a query that only brings back specified columns and first five rows. And you can see the results below.<\/p>\n
df.select(\"userid\", \"teamlevel\").show(5)<\/pre>\n
\"\"<\/p>\n
Next we can use the filter method on the data frame, similar to the\u00a0WHERE<\/strong> clause in a traditional SQL query.<\/p>\n
df.filter(df[\"teamlevel\"] > 1).select(\"userid\" ,\"teamlevel\").show(5)<\/pre>\n
\"\"<\/p>\n
Aggregation Queries<\/h5>\n
Now we will do some simple aggregation queries.<\/p>\n
Now let’s group the results by the value in the IsHit column.<\/p>\n
df.groupBy(\"IsHit\").count().show()<\/pre>\n
\"\"<\/p>\n
If we want to do additional types of aggregations beyond counting we will need to import the SQL functions from the PySpark library. And then we will take the mean and the sum of the IsHit column.<\/p>\n
from pyspark.sql.functions import *\r\ndf.select(mean(\"IsHit\"), sum(\"IsHit\")).show()<\/pre>\n
\"\"<\/p>\n
Joining Tables<\/h5>\n
The last task for this post is to show how to join tables.<\/p>\n
We will read the adclicks table into a dataframe and join it to the game clicks table. We will join on the userid column.<\/p>\n
Similar to above we will use the jdbc driver to load the dataframe and store it in a data frame object.<\/p>\n
df2 = sqlsc.read.format(\"jdbc\") \\\r\n.option(\"url\", \"jdbc:postgresql:\/\/localhost\/cloudera?user=cloudera\") \\\r\n.option(\"dbtable\", \"adclicks\") \\\r\n.load()\r\n\r\ndf2.printSchema()<\/pre>\n
Here are the schema results<\/p>\n
\"\"<\/p>\n
Now we can use the .join method similar to an inner join in an SQL query and join the two data frames.<\/p>\n
dfMerge = df.join(df2, \"userid\")\r\n\r\ndfMerge.printSchema()<\/pre>\n
The merged schema<\/p>\n
\"\"<\/p>\n
dfMerge.show(7)<\/pre>\n
And now we can take a look at what the merged data frames look like.<\/p>\n
\"\"<\/p>\n
In this post we did quite a few things. Looked at loading tables from a PostgreSQL database, performing some simple SQL queries on the data frames, and joined the data frames together.<\/p>\n
I hope you found the above informative, let me know if you have any questions in the comments below.<\/p>\n
\u2014 michael.data@eipsoftware.com<\/a><\/p>\n
 <\/p>\n
 <\/p>\n
 <\/p>\n","protected":false},"excerpt":{"rendered":"
Spark – SQLContext Today we will look at the SQLContext object from the PySpark library and how you can use it to connect to a local database.\u00a0 In the example below we will: Connect to a local PostgreSQL database and read the contents into a dataframe. Run some simple SQL queries And join two data […]<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"_crdt_document":"","footnotes":""},"categories":[3,4,51],"tags":[28,30,52,53],"series":[],"class_list":["post-261","post","type-post","status-publish","format-standard","hentry","category-python","category-code","category-spark","tag-python","tag-code","tag-spark","tag-hadoop"],"_links":{"self":[{"href":"https:\/\/eipsoftware.com\/musings\/wp-json\/wp\/v2\/posts\/261","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/eipsoftware.com\/musings\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/eipsoftware.com\/musings\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/eipsoftware.com\/musings\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/eipsoftware.com\/musings\/wp-json\/wp\/v2\/comments?post=261"}],"version-history":[{"count":5,"href":"https:\/\/eipsoftware.com\/musings\/wp-json\/wp\/v2\/posts\/261\/revisions"}],"predecessor-version":[{"id":276,"href":"https:\/\/eipsoftware.com\/musings\/wp-json\/wp\/v2\/posts\/261\/revisions\/276"}],"wp:attachment":[{"href":"https:\/\/eipsoftware.com\/musings\/wp-json\/wp\/v2\/media?parent=261"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/eipsoftware.com\/musings\/wp-json\/wp\/v2\/categories?post=261"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/eipsoftware.com\/musings\/wp-json\/wp\/v2\/tags?post=261"},{"taxonomy":"series","embeddable":true,"href":"https:\/\/eipsoftware.com\/musings\/wp-json\/wp\/v2\/series?post=261"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}