Impala implements SQL-92 standard features for queries, with some enhancements from later SQL standards.

Because Hadoop Distributed File System (HDFS) is optimized for bulk insert and append operations, Impala currently doesn’t have OLTP-style Data Manipulation Language (DML) operations such as DELETE or UPDATE. It also does not have indexes, constraints, or foreign keys; data warehousing experts traditionally minimize their reliance on these relational features because they involve performance overhead that can be too much when dealing with large amounts of data.

If you have new raw data files, you use LOAD DATA to move them into an Impala table directory. If you have data in one table that you want to copy into another table, optionally filtering, transforming, and converting in the process, you use an INSERT ... SELECT statement. If there is something not satisfactory about some set of data, you replace entire tables or partitions with an INSERT OVERWRITE statement.

The typical OLTP example of depositing to a bank account and withdrawing at the same time is not really appropriate for a data warehousing context. That’s only one row! Impala is intended to analyze what happens across millions or billions of banking operations, ticket purchases, web page visits, and so on.

Impala only appends or replaces; it never actually updates existing data. In write operations, Impala deals not with one row at a time, but millions of rows through statements such as LOAD DATA and INSERT ... SELECT. Even on a transactional DBMS, this volume of new data can be impractical to roll back.

Rather than deleting arbitrary rows with a DELETE statement, you delete large groups of related rows at once, either through DROP TABLE or ALTER TABLE ... DROP PARTITION. If you make a mistake, the original files are still recoverable from the HDFS trashcan.

Operations performed directly in Impala work like the “autocommit” settings available on some database systems. All Impala nodes in the cluster are notified about new data from a LOAD DATA or INSERT statement, or DDL operations such as CREATE TABLE and DROP TABLE.

Remember that Impala has the flexibility to operate on data produced by external programs and pipelines too. When new files are deposited in an Impala table directory by some non-Impala command, the Impala REFRESH table_name statement acts much like a traditional COMMIT statement, causing Impala to re-evaluate the data in that table at the current moment.

Early releases of Impala included binary-style numeric types: 8-bit, 16-bit, 32-bit, and 64-bit integer types, and 32-bit and 64-bit IEEE-754-style floating-point types. These types are well-suited to the kinds of scientific processing and numerical analysis done by many of the Impala early adopters. The following example shows queries with the basic integer and floating-point types.

-- STORE_ID is a SMALLINT, a 32-bit integer that holds up to 32,767.
SELECT DISTINCT(store_id) FROM sales_data;
-- DEGREES is a DOUBLE, a floating-point number from a sensor.
SELECT cos(degrees) FROM telemetry_data;

Impala 1.4 adds support for the DECIMAL data type, which represents base 10 values with varying numbers of significant digits. This type is well-suited for currency calculations, opening up Impala for a range of use cases with financial data. You can also use it to hold integer values with a larger range than the INT or even BIGINT types.

This example shows how with a DECIMAL value (in this case, three digits after the decimal point and nine digits total precision), you get back exactly the value you started with. For some fractional values, FLOAT and DOUBLE are very close but cannot represent them precisely. The extra predictability and accuracy during mathematical operations makes DECIMAL convenient for columns used with GROUP BY, comparisons to literal values, summing large numbers of values, and other cases where the inexact fractional representation of FLOAT and DOUBLE could cause problems.

CREATE TABLE dec_vs_float (dec DECIMAL(9,3), flt FLOAT, dbl DOUBLE);
INSERT INTO dec_vs_float VALUES (98.6,cast(98.6 AS FLOAT),98.6);
SELECT * FROM dec_vs_float;
+--------+-------------------+-------------------+
| dec    | flt               | dbl               |
+--------+-------------------+-------------------+
| 98.600 | 98.59999847412109 | 98.59999999999999 |
+--------+-------------------+-------------------+

Impala’s SQL functionality grows with each release. Here are some of the high points from the Impala 1.4 (July 2014) release. Because this book was finalized well in advance of the Impala 2.0 release, the first edition doesn’t include examples of those new features. Always check the Impala new features documentation page to see what SQL enhancements (along with other kinds of features) were added recently.

Early Impala releases required intermediate results for ORDER BY queries to fit in memory. In terms of syntax, all ORDER BY queries had to also include a LIMIT clause to cap the size of the result set. Impala 1.4 lifts this restriction, saving intermediate sort results to a disk scratch area when necessary.

The DECIMAL data type, introduced in Impala 1.4, lets Impala represent currency data with the kind of accuracy and rounding characteristics that are ideal for financial analysis. See Numbers for more on the DECIMAL type.

Each release of Impala includes additional built-in functions, particularly for math and date/time operations. Impala 1.4 introduced EXTRACT() and TRUNC() for date/time values, and STDDEV() and VARIANCE() for statistical processing.

Although Impala can work with data of any volume, its performance and scalability shine when the data is large enough that you can’t produce, manipulate, and analyze it in reasonable time on a single server. Therefore, after you do your initial experiments to learn how all the pieces fit together, you very quickly scale up to working with tables containing billions of rows and gigabytes, terabytes, or even larger of total volume. The queries that you tinker with might involve data sets bigger than you ever used before.

You might have to rethink your benchmarking techniques if you’re used to using smaller volumes, meaning millions of rows or a few tens of gigabytes. You’ll start relying on the results of analytic queries because the scale will be bigger than you can grasp through your intuition. You’ll become used to adding a LIMIT clause to many exploratory queries to prevent unexpected huge result sets.

For problems that do not tax the capabilities of a single machine, many alternative techniques offer about the same performance. After all, if all you want to do is sort or search through a few files, you can do that plenty fast with Perl scripts or Unix commands such as grep. The Big Data issues come into play when the files are too large to fit on a single machine, when you want to run hundreds of such operations concurrently, or when an operation that takes only a few seconds for megabytes of data takes hours or even days when the data volume is scaled up to gigabytes or petabytes.

You can learn the basics of Impala SQL and confirm that all the prerequisite software is configured correctly using tiny data sets, as in most examples in Chapters 1-4. That’s what we call a canary test, to make sure all the pieces of the system are hooked up properly.

To start exploring scenarios involving performance testing, scalability, and multinode cluster configurations, you typically use much, much larger data sets. Later on, in Tutorial: The Journey of a Billion Rows, we’ll generate a billion rows of synthetic data. Then when the raw data is in Impala, we’ll experiment with different combinations of file formats, compression codecs, and partitioning schemes. We’ll even try some join queries involving a million billion combinations.

Don’t put too much faith in performance results that involve only a few megabytes or gigabytes of data. Only when you blow past the data volume that a single server could reasonably handle, or saturate the I/O channels of your storage array, can you fully appreciate the performance increase of Impala over competing solutions and the effects of the various tuning techniques. To really be sure, do trials using volumes of data similar to your real-world system.

If today your data volume is not at this level, next year it might be. Don’t wait until your storage is almost full (or even half full) to set up a big pool of HDFS storage on cheap commodity hardware. Whether or not your organization has already adopted the Apache Hadoop software stack, experimenting with Cloudera Impala is a valuable exercise to future-proof your enterprise.

Because an HDFS data block contains up to 128 MB by default, you can think of any table less than 128 MB as small (tiny, even). That data could be represented in a single data block, which would be processed by a single core on a single server, with no parallel execution at all. In a partitioned table, the data for each partition is physically split up. Therefore, a table partition of less than 128 MB is in the same situation with limited opportunity for parallel execution. It’s true that the 128 MB block might be split into several smaller files that are processed in parallel. Still, with such small amounts of data, it’s hardly worth the overhead to send the work to different servers across the cluster.

When it comes to Parquet files, Impala writes data files with a default block size of 1 GB. This design choice means that Impala is perfectly happy to process tables or even partitions with many, many gigabytes. For example, if you have a 100-node cluster with 16 cores per node, Impala could potentially process 1.6 TB of Parquet data in parallel, if nothing else were running on the cluster. Larger data volumes would only require a little waiting for the initial set of data blocks to be processed.

Because many organizations do not have those kinds of data volumes, you can decrease the block size before inserting data into a Parquet table. This technique creates a greater number of smaller files. You still want to avoid an overabundance of tiny files, but you might find a sweet spot at 256 MB, 128 MB, 64 MB, or even a little smaller for the Parquet block size. The key is to have enough data files to keep the nodes of the cluster busy, without those files being so small that the overhead of parallelizing the query swamps the performance benefit of parallel execution.

Data stored in Impala is stored in HDFS, a distributed filesystem mounted on one or more Linux servers. When a file is stored in HDFS, the underlying filesystem takes care of making it available across the cluster. Each data block within the file is automatically replicated to some number of hosts (typically at least three), so that all the data is still retrievable if one or two machines experience a hardware, software, or network problem. And when a block needs to be read and processed, that work can be farmed out to any of the servers that hold a copy of that block.

HDFS data blocks are much, much larger than you might have encountered on previous systems. The HDFS block size is typically in the tens of megabytes—often 128 MB or 64 MB. This size is more like what you see with data warehouse software or dedicated analytic hardware appliances. HDFS avoids the issue of wasteful writes by being an append-only filesystem. By reducing the number of possible operations, it focuses on doing a few things well: speed, reliability, and low cost.

I’m always of two minds when talking about text format. It’s the most familiar and convenient for beginners. It’s the default file format for CREATE TABLE commands. It’s very flexible, with a choice of delimiter characters. If you download freely available data sets, such as from a government agency, the data is probably in some sort of text format. You can create textual data files with a simple Unix command, or a Perl or Python script on any computer whether or not it’s running Hadoop. You can fix format errors with any text editor. For small volumes of data, you can even do your own searches, sorts, and so on with simple Unix commands like grep, awk, and sort. Within Impala, you can change your mind at any time about whether a column is a STRING, INT, TINYINT, BIGINT, and so on. One minute you’re summing numbers, the next you’re doing SUBSTR() calls to check for leading zeros in a string, as illustrated in Numbers Versus Strings.

And yet, it’s also the bulkiest format, thus the least efficient for serious Big Data applications. The number 1234567 takes up 7 bytes on disk; –1234567 takes up 8 bytes; –1234.567 takes up 9 bytes. The current date and time, such as 2014-07-09 15:31:01.409820000, takes up 29 bytes. When you’re dealing with billions of rows, each unnecessary character represents gigabytes of wasted space on disk, and a proportional amount of wasted I/O, wasted memory, and wasted CPU cycles during queries.

Therefore, I’m going to advise again and again to prefer Parquet tables over text tables wherever practical. The column-oriented layout and compact storage format, with compression added on top, make Parquet the obvious choice when you are dealing with Big-Data-scale volume.

These examples demonstrate creating tables for text data files. Depending on the format of the input data, we specify different delimiter characters with the rather lengthy ROW FORMAT clause. STORED AS TEXTFILE is optional because that is the default format for CREATE TABLE. The default separator character is hex 01 (ASCII Ctrl-A), a character you’re unlikely to find or enter by accident in textual data.

CREATE TABLE text_default_separator
  (c1 STRING, c2 STRING, c3 STRING);

CREATE TABLE text_including_stored_as_clause
  (c1 STRING, c2 STRING, c3 STRING) STORED AS TEXTFILE;

CREATE TABLE csv (c1 STRING, c2 STRING, c3 STRING)
  ROW FORMAT DELIMITED FIELDS TERMINATED BY "," STORED AS TEXTFILE;

CREATE TABLE tsv (c1 STRING, c2 STRING, c3 STRING)
  ROW FORMAT DELIMITED FIELDS TERMINATED BY "	" STORED AS TEXTFILE;

CREATE TABLE psv (c1 STRING, c2 STRING, c3 STRING)
  ROW FORMAT DELIMITED FIELDS TERMINATED BY "|" STORED AS TEXTFILE;

The Parquet file format, which originated from collaboration between Twitter and Cloudera, is optimized for data-warehouse-style queries. Let’s explore what that means and how it affects you as a developer.

Parquet is a binary file format. Numeric values are represented with consistent sizes, packed into a small number of bytes (either 4 or 8) depending on the range of the type. TIMESTAMP values are also represented in relatively few bytes. BOOLEAN values are packed into a single bit, rather than the strings true and false as in a text table. So all else being equal, a Parquet file is smaller than the equivalent text file.

But Parquet has other tricks up its sleeve. If the same value is repeated over and over, Parquet uses run-length encoding to condense that sequence down to two values: the value that’s repeated, and how many times it’s repeated. If a column has a modest number of different values, up to 16K, Parquet uses dictionary encoding for that column: it makes up numeric IDs for the values and stores those IDs in the data file along with one copy of the values, rather than repeating the values over and over. This automatically provides space reduction if you put denormalized data straight into Parquet. For example, if a data file contains a million rows, and each has one column with a state name such as California or Mississippi, the data file is essentially the same whether you convert those strings to state #1, #2, and so on and store the numbers, or if you let Parquet’s dictionary encoding come up with the numeric IDs behind the scenes. The limit of 16K distinct values applies to each data file, so if your address table has more than 16K city names, but the table is partitioned by state so that the California cities are in one data file and the Mississippi cities are in a different data file, each data file could still use dictionary encoding for the CITY column.

See Tutorial: The Journey of a Billion Rows for a demonstration of how much space you can save with Parquet format when all the compression and encoding techniques are combined.

Parquet is a column-oriented format. This means that the data for all values in a column are arranged physically adjacent to each other on disk. This technique speeds up certain kinds of queries that do not need to examine or retrieve all the columns, but do need to examine all or most of the values from particular columns:

-- The query can read all the values of a column without having to
-- read (and ignore) the values of the other columns in each row.
SELECT c3 FROM t1;

-- Analytic queries are always counting, summing, averaging and so on
-- columns for sales figures, web site visitors, sensor readings, and so on.
-- Those computations are nice and fast when no unnecessary data is read.
-- In this example, the query only needs to read C1 and C5, skipping all
-- other columns.
SELECT count(DISTINCT c1), sum(c1), max(c1), min(c1), avg(c1)
  FROM t1 WHERE c5 = 0;

-- Here we cross-reference columns from two different tables, along
-- with an ID column that is common to both. Again, the query only reads
-- values from the exact columns that are needed, making join queries
-- practical for tables in the terabyte and petabyte range.
SELECT attr01, attr15, attr37, name, email FROM
  visitor_details JOIN contact_info USING (visitor_id)
  WHERE year_joined BETWEEN 2005 AND 2010;

Column-oriented data is a popular feature in specialized data warehouse systems. For Impala, the column-oriented file format is just a small piece of the value proposition. The file format itself is open, so you can always get data in or out of it. Parquet files are readable and writable by many Hadoop components, so you can set up an ETL pipeline to use Parquet all the way through rather than starting in one format and converting to another at the end.

People commonly assume that the Parquet column-oriented format means each column is stored in a different data file. Not so! Each Parquet data file contains all the columns for a group of rows, but the values from each column are laid out next to each other within that file. When Impala needs to read all the values from a particular Parquet column, it seeks to a designated spot in the file and reads forward from there. The performance benefits of Parquet increase as you add more columns; for example, with 100 columns, a query only needs to read roughly 1% of each data file for each column referenced in the query.

The SHOW TABLE STATS statement provides the basic information about the file format of the table, and each individual partition where appropriate:

[localhost:21000] > show table stats csv;
+-------+--------+------+--------------+--------+
| #Rows | #Files | Size | Bytes Cached | Format |
+-------+--------+------+--------------+--------+
| -1    | 1      | 58B  | NOT CACHED   | TEXT   |
+-------+--------+------+--------------+--------+

The DESCRIBE FORMATTED statement dumps a lot of information about each table, including any delimiter and escape characters specified for text tables:

[localhost:21000] > describe formatted csv;
...
| Storage Desc Params: | NULL                  | NULL |
|                      | field.delim           | ,    |
|                      | serialization.format  | ,    |
+----------------------+-----------------------+------+

Your preferred file format might change over time, after you conduct benchmarking experiments, or because of changes earlier in your ETL pipeline. Impala preserves the flexibility to change a table’s file format at any time: simply replace the data with a new set of data files and run an ALTER TABLE ... SET FILEFORMAT statement. Or, for a partitioned table, you can leave older partitions in the previous file format, and use the new file format only for newer partitions.

This example demonstrates how to clone the structure of an existing table, switch the file format of the new table, and then copy data from the old to the new table. The data is converted to the new format during the copy operation.

CREATE TABLE t2 LIKE t1;
-- Copy the data, preserving the original file format.
INSERT INTO t2 SELECT * FROM t1;
ALTER TABLE t2 SET FILEFORMAT = PARQUET;
-- Now reload the data, this time converting to Parquet.
INSERT OVERWRITE t2 SELECT * FROM t1;

The following example demonstrates how a partitioned table could start off with one file format, but newly added partitions are switched to a different file format. Queries that access more than one partition automatically accommodate the file format for each partition.

CREATE TABLE t3 (c1 INT, c2 STRING, c3 TIMESTAMP)
  PARTITIONED BY (state STRING, city STRING);
ALTER TABLE t3 ADD PARTITION
  (state = 'CA', city = 'San Francisco'),
-- Load some text data into this partition...
ALTER TABLE t3 ADD PARTITION
  (state = 'CA', city = 'Palo Alto'),
-- Load some text data into this partition...
ALTER TABLE t3 ADD PARTITION
  (state = 'CA', city = 'Berkeley'),
ALTER TABLE t3 PARTITION
  (state = 'CA', city = 'Berkeley')
  SET FILEFORMAT = PARQUET;
-- Load some Parquet data into this partition...