ETL with Hive

Extract, transform, load

When you have existing data you want to load into Hive without creating a link to the source data by using an external table, Hive offers many options. In fact, Hive has several commands to support ETL, all of which result in populating an internal Hive table.

You can populate Hive from a subset of data from HBase and split the row key into parts for indexing, or you can load files from either HDFS or the local file system. As part of the load, you can transform the data into more usable representations, or you can strip out only the parts you need.

Hive offers multiple options for ETL, but all of the processing is done through map/reduce jobs, which means loading data of any size is possible. Hive is particularly well suited for the alternative data input approach ETL because it supports very efficient loading of data in the native format, which can then be transformed by reading and writing Hive tables.

In this chapter we’ll cover the major commands for getting data into Hive in the format of your choosing.

Loading from files

The simplest ETL tool is the load command. Simply put, it loads a file or set of files into an existing internal Hive table. Running load is suitable only when the data in the source files matches the target table schema, because you cannot include any transformations in this option.

You should also know that the load statement will not do any verification, which means you can load files with the wrong format into a table and the command will run, but you won’t be able to read the data back again.

With load you can specify the source file(s) using either a local filepath or an HDFS path. Code Listing 47 shows an example that loads a syslog file from the local /tmp directory into the syslogs_flat table and then fetches the first row.

Code Listing 47: Loading Data into Hive

The load command has two qualifiers for changing its behavior:

• LOCAL—specifies that the source filepath is on the local machine. If omitted, the filepath is assumed to be HDFS.
• OVERWRITE—deletes the existing contents of the table before loading the new files. If omitted, the new data is appended to the table.

Code Listing 48 shows an alternative use of the load command—appending the existing data in the syslogs_flat table from a file in HDFS with counts before and after the load in order to show the number of rows in the table.

Code Listing 48: Appending Data to Hive

Note: the source file path is specified as a full URI here in order to show that a remote HDFS cluster can be used, but you can also specify relative or absolute paths for files in the home HDFS cluster used by Hive.

The load command can only be used with internal tables because its single function is to copy files from the specified source to the underlying folder in HDFS for the Hive table. Code Listing 49 shows the contents of the table folder after the two preceding load operations.

Code Listing 49: Listing Files Loaded into Hive

With the overwrite flag, Hive deletes the existing files before copying the new source files. Code Listing 50 shows the result of overwriting the table with a new file and also shows the HDFS file listing after the load.

Code Listing 50: Load with Overwrite

The load statement is a powerful and quick way of loading data into Hive because it simply does an HDFS put to copy the files from the source into the Hive table structure. However, its use is limited to cases in which the source files are in the correct format for the table and no transformation can occur as part of the load.

Note: Hive also has the import and export functions that let you save or load tables to an HDFS location. These are are different from load, in that the Hive metadata gets exported and imported along with the data, so that you can use those functions effectively to back up a table on one Hive instance and recreate it on another.

We can see the limitations of having untransformed data in the syslogs_flat table. For example, syslog files from a Linux machine can be easily loaded, but the format is not easily mapped in Hive. Each entry uses space-separated fields for date, machine name, and event type, along with a colon before the message. Data can’t be transformed as part of a load, so I have a Hive table with a single string column in which each row contains all the log entry data in one string value.

The rows in the syslogs_flat table aren't well suited for querying if we use load, but now that the data is in Hive we can use other options to transform the data into a more usable format and load it into other tables.

This is more of an extract, load, transform (ELT) approach in which data is first loaded into Hive in its native format, which is fast, and then transformed. Transformation can be scheduled as map/reduce jobs by Hive.

Inserting from query results

Hive supports loading a table with the results of a query on other objects, and you can include transformation functions in the query. You can also populate internal or external tables this way, with the query parser doing some basic validation in order to ensure that the columns from the source query will fit into the target table.

Columns are positionally matched between source and target, which means you need to craft your select statement so that the columns in the result set are in the same order as the columns defined in the target table.

If there are too many or too few columns in the query, Hive won't try to match them to the target table and you'll get an error, as in Code Listing 51.

Code Listing 51: Invalid Insert Statement

Tip: You must include a from clause in an insert…select statement, which means you can't use literal expressions such as select 1, 'a'. But you can create a table with one row and one column (I name it dual after the convention in Oracle databases), and then you can use dual in the from clause. You will find that insert into [table] select 'a', 'b', 3 will fail, but insert into [table] select 'a', 'b', 3 from dual will succeed.

Hive doesn't verify the data types of the column, which means you can load data into the wrong columns without any errors. The type mismatch will manifest itself when you attempt to query the data and Hive can't map the values, so that the results contain nulls.

The query clause can contain any valid HiveQL, including joins and unions between internal and external tables, function calls, and aggregation, which allows for a lot of extract and transform logic. In order to make a set of syslog files more usable, I've defined a new table with separate columns for the data items, as in Code Listing 52.

Code Listing 52: A Structured Table for Syslogs

We'll see more HiveQL functions in Chapter 9  Querying with HiveQL, and there is a useful string function called sentences that takes an input string and returns it tokenized as an array of sentences, each containing an array of words. I can use that to pull out specific words from the log entry, as in Code Listing 53, in which I also cast strings to other types.

Code Listing 53: Splitting String Fields

The timestamp of the log entry is a little trickier, especially because Ubuntu doesn't record the year in the log, but if we assume the current year we can use a combination of string and date functions to prepend the year to the date and time in the entry, then convert it all to a timestamp, as in Code Listing 53.

Code Listing 54: Transforming Strings to Timestamps

The final column is the log message, which is a straightforward substring after the closing square bracket from the process ID, as in Code Listing 55.

Code Listing 55: Extracting Substrings

Putting it all together, we can populate the new syslogs table from the raw syslogs_flat table with a single insert … select. The query aspect is clunky because of the nature of the input data, but once it runs we can make much more useful selections over the formatted syslogs table, as in Code Listing 56.

Code Listing 56: Transforming Raw Data

Inserting query results to tables also supports the overwrite clause, which effectively truncates the target table before inserting the results of the query.

Multiple inserts

Hive supports an extended version of insert … select in which multiple insert statements from multiple queries over the same source can be chained. In this variation, you would begin by specifying the source table (or view), then adding the inserts.

Multiple inserts are useful when you need to populate different Hive projections from a single source, because multiple inserts run very efficiently. Hive will scan the source data once, then run each query over the scanned data, inserting it into the target.

In Code Listing 57 we use the unformatted syslogs table as the source and load the formatted table and a summary table at the same time.

Code Listing 57: Multiple Inserts

Create table as select

A final variation on inserting data from query results is create table as select (CTAS), which lets us define a table and populate it with a single statement. The table can only be internal, but we can specify the normal stored by clauses for internal tables.

In a CTAS table, the structure is inferred from the query, which means we don't need to explicitly specify the columns. Queries should cast values into the data types we want for the target table and should specify aliases that will be used as the column names.

The CTAS operation is atomic, which means the table will not appear as available for querying until the CTAS completes and the table is populated.

Code Listing 58 shows a CTAS statement for loading the individual words from syslog entries into a new table. The select statement parses the log message as sentences and extracts the first sentence as an array of strings, which is how Hive defines the column.

Code Listing 58: Create Table as Select

Temporary tables

Hive supports temporary tables, which are very useful for interim data transformations in ETL/ELT workloads that cannot achieve the full transform in a single step. Temporary tables don't support indexes, and they exist only for the life of the Hive session—they are automatically deleted when the session ends.

Temporary tables are internal tables stored for the user in the working directory within HDFS. They can be specified with a supported file format, so that you benefit from efficient storage while you use the table. However, note that not all functionality is supported.

Code Listing 59 shows a temporary table being created to store the progress of an ETL job.

Code Listing 59: Using Temporary Tables

In this case the insert statement uses a literal date because Hive doesn't support using functions in the values clause for inserting into a temporary table. If you try using a built-in function, such as unix_timestamp, in order to get the current time, you'll receive an error.

However, you can use functions in a select clause, which means you can use the same trick of selecting literals from dual, as shown in Code Listing 60.

Code Listing 60: Inserting from Functions

Whether the session is interactive with Beeline or a job submitted through an external interface, Hive deletes the table at the end of the session. The temporary table is visible only in the session that created it. Other sessions, even for the same user, will not see the table.

When the session that created the temporary table ends, the data and metadata for the table are removed, and when the next session starts the table will be gone, as we see in Code Listing 61.

Code Listing 61: Temporary Tables Being Removed

In this example, when the Beeline user disconnects with the !close command, the session ends, and the Hive server deletes the temporary table. When the user connects again, a new session begins, and the temporary table will be gone.

Summary

With the basic load, insert, and CTAS statements, Hive supports the major patterns for getting data into the warehouse. If you have existing processes for extracting and transforming data, you can load those directly into Hive tables. That fast operation will result in data being securely stored in HDFS and available for querying through Hive.

For new data loads, an ELT process makes better use of Hive by initially using load to put the raw data into Hive tables and transforming it using HiveQL. The insert … select and create table … as select statements allow you to craft a complex query with functions to transform your data and have it populated in Hive through scalable map/reduce jobs.

Once you have the data as tables in Hive, you can create views and indexes to make the information approachable. When your next set of data is extracted, simply repeat the inserts and the new data will be appended to the existing tables.

In the next chapter we’ll look more closely at defining objects and modifying data using HiveQL, the Hive Query Language.