Now our job flow is ready so lets SSH into the master node.
The command line client provides an easy way to do this, just specify the SSH option and
give it a job flow ID.
The command runs SSH and soon you'll be connected to the master node.
Type Hive to start the Hive command line interface.
From here you'll be able to create tables and run queries.
Our tables are stored in JSON format. In order to read them, we'll need to use a custom JSON
SerDe.
SerDe is a feature of Hive and stands for Serializer/Deserializer.
They are written in Java and used to read custom data formats.
To use the SerDe, we need to make Hive aware of it by calling "add jar". The location should
be a path in S3.
This specific SerDe was written by Amazon and can read JSON files.
Now, let's create some tables.
First we can type "show tables" and see that none currently exist.
I am going to continue with the advertising company story.
We have two tables that are published by our ad servers.
One that logs each impression and one that logs each ad click.
Every five minutes the servers roll their logs and upload them to S3.
The tables are partitioned by time, so the ad servers have to stick the files in a specific
directory with other logs from that time period.
In order to read the table, Hive must know about both the partition format and the table
SerDe.
Let's go ahead and create the impressions table.
You'll notice the create table statement is similar to those in relational databases,
but with a few custom parameters.
This is an external table, which means that Hive is aware of the table, but doesn't manage
the underlying storage itself.
I specify "if not exists" so that Hive won't delete the table if it already exists.
I'll define each of the columns and tell Hive about their type.
This table is partitioned. By default, Hive overwrites the table upon insert, but if it
is partitioned, then it can overwrite just one partition, allowing you to perform inserts.
Hive can also use the partition information to limit the amount of data it scans if you
include the partition in the where clause.
This table is partitioned with a date time string, which we must include in the create
table statement.
The files in our table contain one JSON object per line.
In order for Hive to read the table, we need to tell it to use our custom SerDe.
Since we already included the jar file containing the SerDe, now we can reference it by class
name.
You can configure the SerDe with the SerDe properties statement.
Our SerDe takes a path option which allows us to extract items from the JSON object by
specifying their path.
Each path corresponds to a column above.
Finally, we point to the location of our table in S3.
Hit enter and our table will be created.
Hive keeps track of table partitions in its metadata store.
Right now, Hive does not know about any of our partitions because the table was just
created.
We could inform Hive of the partitions one by one, but this would be time consuming.
Fortunately, we added a feature to Hive that lets you recover partitions.
If you type "alter table recover partitions", Hive will scan the table location and discover
all the partitions automatically.
If we type "show partitions" again, we can see all partitions it discovered.
To prove to ourselves that the table is configured properly, we can perform a select * on the
table.
This will print all the results to the screen. Since the table is large, we'll limit it to
ten rows.
Everything looks okay.
I'd also like to point out that the last column is the partition name.
If you filter on this in the where clause, Hive will only read from the necessary files
and won't perform a full table scan.
Now lets go ahead and create our other table which includes information on ad clicks.
This is a similar process to creating the impressions table, but clicks has fewer columns.
Don't forget to include the partition information and JSON SerDe.
This table is also stored in Amazon S3.
Since this table was just created, Hive doesn't know about its partitions.
We can use recover partitions again to automatically add them.
Let's make sure that worked correctly by listing the partitions again.
Yep, everything looks okay.
Now that we are done with our input tables, it is time to create the output table.
This will contain the result of joining the impressions and clicks tables.
We will call this table joined_impressions.
It contains all the columns from impressions, plus a new column called clicked.
Clicked will be set to true if that impression resulted in an ad click.
This table will be partitioned on two values: day and hour.
This allows us to schedule an hourly job which takes the last hours worth of data from our
input tables and inserts it into this table.
Instead of using JSON, this table will be stored on disk in sequence file format.
Sequence file is a storage container that ships with Hadoop and offers better performance
and more compression than JSON.
Finally, we point to the location in S3 where we want to store our table.
This table does not contain any data yet, so there is no need to recover partitions.
Now lets go ahead and create the query used data into this table.
We begin with a simple "insert overwrite table".
By default, this will erase all data in the table. However, our table is partitioned,
so we can overwrite just a single partition.
When inserting into a partitioned table, we must tell Hive which partition we are writing
to with the partition statement.
This statement takes a value for each of the table's partitions, which in our case is day
and hour.
We have told Hive where to put the data, now we need to tell it where to read the data
using a select statement.
As Hive joins the tables, we would also like to perform a transformation on request begin
time and convert it from a unix timestamp into a human readable date.
This can be done easily using some of Hive's built in functions.
Hive supports user defined functions so you can create your own as well.
We will be joining the tables using a left outer join. The results will include every
impression from our time period.
If the impression resulted in an ad click, then the corresponding row from the clicks
table will be included with the impression, otherwise the click will be null.
This means we can set our clicked boolean by testing if the clicked row is null.
We tell Hive which tables to read from and tell it to join them on impression ID.
Finally, we limit our query to a certain time range using the where clause.
Since we are filtering on the partition column, Hive should read only the minimum number of
files necessary to retrieve our data.
You may notice the time range we are using for the clicks table extends twenty minutes
longer than that of the impressions table.
This is to take into account that clicks don't occur at the same time as the impression.
Our software accepts clicks that occur up to twenty minutes after the impression.
When I hit enter, Hive parses the query and compiles it into a series of MapReduce jobs.
This is a fairly simple query, so Hive is able to perform all the calculations in one
MapReduce pass.
Hive automatically submits your job to the Hadoop cluster and waits for it to finish.
Here we can see the progress of the map and reduce stages.
I've sped up the video to make it go faster.
Once the job has finished, we can look at our data to make sure it looks okay.
Let's select just ten rows so we can get a sense of the data.
Notice how the clicked boolean is set in the table.
Also, you can see the day and hour partitions in the results.
If you look at the very first column in each row, you'll notice that the request begin
time is now in an easy to read format and is no longer a unix timestamp.
I've just demonstrated a data warehouse-like query that joins data from two tables.
I ran this in interactive mode and only processed one hours worth of data.
If you were using this query in production, you would use the Hive batch mode and have
a schedular that automatically submits steps to Elastic MapReduce.
For more Hive examples within this use case, you can read the tutorials on our website
at aws.amazon.com/elasticmapreduce.
Thank you for listening.