Using Apache Kudu (incubating) with Apache Impala (incubating)

If you use parcels, Cloudera recommends using the included deploy.py script to install and deploy the Impala_Kudu service into your cluster. If your cluster does not have an existing Impala instance, the script is optional. However, if you do have an existing Impala instance and want to install Impala_Kudu side-by-side, you must use the script.

You can create a table by querying any other table or tables in Impala, using a CREATE TABLE …​ AS SELECT statement. The following example imports all rows from an existing table old_table into a Kudu table new_table. The columns in new_table will have the same names and types as the columns in old_table, but you need to populate the kudu.key_columns property. In this example, the primary key columns are ts and name.

You can refine the SELECT statement to only match the rows and columns you want to be inserted into the new table. You can also rename the columns by using syntax like SELECT name as new_name.

Tables are divided into tablets which are each served by one or more tablet servers. Ideally, tablets should split a table’s data relatively equally. Kudu currently has no mechanism for automatically (or manually) splitting a pre-existing tablet. Until this feature has been implemented, you must pre-split your table when you create it, When designing your table schema, consider primary keys that will allow you to pre-split your table into tablets which grow at similar rates. You can provide split points using a DISTRIBUTE BY clause when creating a table using Impala:

If you have multiple primary key columns, you can specify split points by separating them with commas within the inner brackets: (('va',1), ('ab',2)). The expression must be valid JSON.

Impala uses a database containment model. In Impala, you can create a table within a specific scope, referred to as a database. To create the database, use a CREATE DATABASE statement. To use the database for further Impala operations such as CREATE TABLE, use the USE statement. For example, to create a table in a database called impala_kudu, use the following statements:

The my_first_table table is created within the impala_kudu database. To refer to this database in the future, without using a specific USE statement, you can refer to the table using <database>.<table> syntax. For example, to specify the my_first_table table in database impala_kudu, as opposed to any other table with the same name in another database, use impala_kudu.my_first_table. This also applies to INSERT, UPDATE, DELETE, and DROP statements.

The following Impala keywords are not supported when creating Kudu tables: - PARTITIONED - STORED AS - LOCATION - ROWFORMAT

You can specify split rows for one or more primary key columns that contain integer or string values. Range partitioning in Kudu allows splitting a table based based on the lexicographic order of its primary keys. This allows you to balance parallelism in writes with scan efficiency.

The split row does not need to exist. It defines an exclusive bound in the form of: (START_KEY, SplitRow), [SplitRow, STOP_KEY) In other words, the split row, if it exists, is included in the tablet after the split point. For instance, if you specify a split row abc, a row abca would be in the second tablet, while a row abb would be in the first.

Suppose you have a table that has columns state, name, and purchase_count. The following example creates 50 tablets, one per US state.

Instead of distributing by an explicit range, or in combination with range distribution, you can distribute into a specific number of 'buckets' by hash. You specify the primary key columns you want to partition by, and the number of buckets you want to use. Rows are distributed by hashing the specified key columns. Assuming that the values being hashed do not themselves exhibit significant skew, this will serve to distribute the data evenly across buckets.

You can specify multiple definitions, and you can specify definitions which use compound primary keys. However, one column cannot be mentioned in multiple hash definitions. Consider two columns, a and b: * HASH(a), HASH(b) * HASH(a,b) * HASH(a), HASH(a,b)

Hash partitioning is a reasonable approach if primary key values are evenly distributed in their domain and no data skew is apparent, such as timestamps or serial IDs.

The following example creates 16 tablets by hashing the id column. This spreads writes across all 16 tablets. In this example, a query for a range of sku values is likely to need to read all 16 tablets, so this may not be the optimum schema for this table. See Advanced Partitioning for an extended example.

You can combine HASH and RANGE partitioning to create more complex partition schemas. You can specify zero or more HASH definitions, followed by zero or one RANGE definitions. Each definition can encompass one or more columns. While enumerating every possible distribution schema is out of the scope of this document, a few examples illustrate some of the possibilities.

This example creates 100 tablets, two for each US state. Per state, the first tablet holds names starting with characters before 'm', and the second tablet holds names starting with 'm'-'z'. Writes are spread across at least 50 tablets, and possibly up to 100. A query for a range of names in a given state is likely to only need to read from one tablet, while a query for a range of names across every state will likely read from at most 50 tablets.

Consider the simple hashing example above, If you often query for a range of sku values, you can optimize the example by combining hash partitioning with range partitioning.

The following example still creates 16 tablets, by first hashing the id column into 4 buckets, and then applying range partitioning to split each bucket into four tablets, based upon the value of the sku string. Writes are spread across at least four tablets (and possibly up to 16). When you query for a contiguous range of sku values, you have a good chance of only needing to read from a quarter of the tablets to fulfill the query.

Again expanding the example above, suppose that the query pattern will be unpredictable, but you want to ensure that writes are spread across a large number of tablets You can achieve maximum distribution across the entire primary key by hashing on both primary key columns.

The example creates 16 buckets. You could also use HASH (id, sku) INTO 16 BUCKETS. However, a scan for sku values would almost always impact all 16 buckets, rather than possibly being limited to 4.

In general, be mindful the number of tablets limits the parallelism of reads, in the current implementation. Increasing the number of tablets significantly beyond the number of cores is likely to have diminishing returns.

This example inserts a single row.

This example inserts three rows using a single statement.

When inserting in bulk, there are at least three common choices. Each may have advantages and disadvantages, depending on your data and circumstances.

Normally, if you try to insert a row that has already been inserted, the insertion will fail because the primary key would be duplicated. See Failures During INSERT, UPDATE, and DELETE Operations. If an insert fails part of the way through, you can re-run the insert, using the IGNORE keyword, which will ignore only those errors returned from Kudu indicating a duplicate key..

The first example will cause an error if a row with the primary key 99 already exists. The second example will still not insert the row, but will ignore any error and continue on to the next SQL statement.

You can update in bulk using the same approaches outlined in Inserting In Bulk.

Similarly to INSERT and the IGNORE Keyword, you can use the IGNORE operation to ignore an UPDATE which would otherwise fail. For instance, a row may be deleted while you are attempting to update it. In Impala, this would cause an error. The IGNORE keyword causes the error to be ignored.

You can delete in bulk using the same approaches outlined in Inserting In Bulk.

Similarly to INSERT and the IGNORE Keyword, you can use the IGNORE operation to ignore an DELETE which would otherwise fail. For instance, a row may be deleted by another process while you are attempting to delete it. In Impala, this would cause an error. The IGNORE keyword causes the error to be ignored.