New Features in CDH 6.0.0

YARN jobs running MapReduce2 are now configured to read MapReduce JARs from HDFS instead of from local disk. This makes jobs more robust during rolling upgrades when the local binaries are modified while a job is executing. Clusters created in or upgrading to CDH 6.0 will use this new behavior.

CDH 6.0.0 has some significant architectural changes to the assignment manager. The assignment manager is a component in the HBase master that manages region to region server assignment and ensures that any one region replica is assigned to just one region server. The overall goal is to make HBase more resilient through providing a unified place within HBase to handle some tricky scenarios related to region assignment including:

Note that HBCK is read-only in CDH 6.0.0. HBASE-18792 is listed in the CDH HBase Incompatible Changes.

Replication peers can be defined for entire namespaces instead of for each individual table. For clusters with hundreds or thousands of tables defining replication configuration settings for each table is cumbersome and error prone and being able to define configuration at the namespace level is a significant improvement.

Periodically, the HBase master will look at load across region servers and attempt to balance it out by reassigning regions. The handling of this cluster maintenance has improved in CDH 6.0.0:

HBase’s command line shell has been enhanced with several new commands that allow for easier cluster administration. New commands allow for administrators to better understand the current compaction state of a table and to clear the compaction queue if needed. There is also a new command to display metadata associated with a region.

HBase deployments with a heavy write workload often disable HBase's automatic handling of major compactions in order to reduce IO during peak cluster load. HBase now ships a tool to make it easier for operators who then have to ensure major compactions can happen during off-hours.

• HBASE-19528: Major Compaction Tool - Tool allows you to compact a cluster with given concurrency of regionservers compacting at a given time.

In CDH 6.0.0, there are a significant number of new metrics available. Additionally, client side metrics are now collected by default. Some examples include:

CDH 6.0.0 includes the ability to define an off-heap write cache. Off-heap caches allow HBase to manage memory directly instead of delegating memory management to java. HBase has much more context on how to safely reclaim memory it’s using compared to the java garbage collection process which allows HBase to more efficiently manage memory usage. Off-heap caching allows HBase to use more memory than would be practical if delegating the responsibility to java where “stop the world” garbage collection can cause large spikes in latency. One consequence of allowing significantly more write side cache memory through an off-heap cache is an increase in density of regions per region server.

It is now possible to configure HBase so that data remains off-heap along the entire read path, avoiding copies of the data onto the java heap. With this feature enabled cached data will have a similar latency to on-heap caching in earlier versions of HBase but without putting GC pressure on the JVM. For more information on configuring and running a deployment with an off-heap read path see the Apache HBase documentation reference guide section on Offheap Read Path.

CDH 6.0.0 has an option to allow HFileCleaner to use multiple threads. HFileCleaner is a background process that removes files containing dereferenced data. This new configuration is useful when a system is under substantial write load to ensure that clean up operations complete in a timely manner to prevent the archive directory from ballooning in size.

An additional performance enhancement in CDH 6.0.0 is the introduction of a “streaming scan” API. This API can improve the performance of intensive workloads by allowing multiple chunks of data to be prefetched concurrently. Previously, prefetch operations were synchronous, i.e, they required the previous prefetch operation to complete. Synchronous prefetches can result in a “stop-and-wait” access pattern where an application has finished processing its previous chunk of data and is now waiting for the next chunk of data to arrive. Prefetching a second chunk of data before the first is entirely processed helps eliminate the “stop-and-wait” access pattern.

A new RPC server based on Netty4 is included in CDH 6.0.0. This RPC server option can increase the performance of workloads with large amounts of random reads and writes. This new RPC server option is on by default.

HBase provides users the ability to run map/reduce jobs over HBase snapshots. Snapshots are immutable, and by running map/reduce jobs directly against them clients can avoid performance overhead associated with going through the region servers. In prior HBase versions exactly one map task per region was created. In CDH 6.0.0, the ability to assign multiple mappers to a particular region is possible. This change can further improve the performance of map/reduce jobs over HBase snapshots and eliminates the need to design schemas optimized for the prior limitation.

HDFS has supported heterogeneous storage management (HSM) for several releases. This feature allows users to specify that certain HDFS files or directories should reside on specific physical media. In CDH 6.0.0, users can do something very similar at the logical, column-family level. Currently supported storage policies include ALL_SSD/ONE_SSD/HOT/WARM/COLD.

Note that HBase has supported setting storage policies on WALs since CDH 5.7.

Clients can now optionally elect to handle slow region servers in application code. By setting the configuration value for hbase.client.perserver.requests.threshold an application can control how many concurrent requests to a single region server are allowed. After reaching this threshold the HBase client code will throw a ServerTooBusyException rather than attempting to establish an additional connection. If an HBase cluster has a small number of region servers that are having trouble this will allow an application to react in some way rather than simply waiting for the server to eventually respond. Because using this feature requires application side changes to handle retry logic you must opt-in to it; by default hbase.client.perserver.requests.threshold is set to practically unlimited.

By default, the Hive query execution engine processes one row of a table at a time. The single row of data goes through all the operators in the query before the next row is processed, resulting in very inefficient CPU usage. In vectorized query execution, data rows are batched together and represented as a set of column vectors. The query engine then processes these vectors of columns, which greatly reduces CPU usage for typical query operations like scans, filters, aggregates, and joins.

Hive query vectorization is enabled by setting the hive.vectorized.execution.enabled property to true. In both CDH 5 and CDH 6, this property is set to true by default. But in CDH 5, vectorized query execution in Hive is only possible on ORC-formatted tables, which Cloudera recommends that you do not use for overall compatibility with the CDH platform. Instead, Cloudera recommends that you use tables in the Parquet format because all CDH components support this format and can be consumed by all CDH components. In CDH 6, query vectorization is supported for Parquet tables in Hive.

For more information, see Query Vectorization for Apache Hive in CDH and Apache Parquet Tables with Hive in CDH.

Support has been added for the UNION DISTINCT clause in Hive. See HIVE-9039 and the Apache wiki for more details. This feature introduces the following incompatible changes to Hive:

• Behavior in CDH 5:

  • SORT BY, CLUSTER BY, ORDER BY, LIMIT, and DISTRIBUTE BY can be specified without delineating parentheses either before a UNION ALL clause or at the end of the query, resulting in the following behaviors:

    • When specified before, these clauses are applied to the query before UNION ALL is applied.
    • When specified at the end of the query, these clauses are applied to the query after UNION ALL is applied.
  • The UNION clause is equivalent to UNION ALL, in which no duplicates are removed.

• Behavior in CDH 6:

  • SORT BY, CLUSTER BY, ORDER BY, LIMIT, and DISTRIBUTE BY can be specified without delineating parentheses only at the end of the query, resulting in the following behaviors:

    • These clauses are applied to the entire query.
    • Specifying these clauses before the UNION ALL clause results in a parsing error.
  • The UNION clause is equivalent to UNION DISTINCT, in which all duplicates are removed.

Support has been added for NULLS FIRST and NULLS LAST options. These options can be used to determine whether null values appear before or after non-null data values when the ORDER BY clause is used. Hive follows the SQL:2003 standard for this feature, but the SQL standard does not specify the behavior by default. By default in Hive, null values are sorted as if lower than non-null values. This means that NULLS FIRST is the default behavior for ASC order, and NULLS LAST is the default behavior for DESC order. See Syntax of Order By on the Apache Hive wiki and HIVE-12994 for further details.

Here are some usage examples:

SELECT x.* FROM table1 x ORDER BY a ASC NULLS FIRST;
SELECT x.* FROM table1 x ORDER BY a ASC NULLS LAST;
        

Support for the following has been added to CDH 6.0:

• Using DISTINCT with windowing functions. See HIVE-9534 for details.
• Support for ORDER BY and a windowing clause when DISTINCT is used in a partitioning clause. See HIVE-13453 for details.
• Support to reference aggregate functions within the OVER clause. See HIVE-13475 for details.

For further details, see the Apache Language Manual on Windowing and Analytics.

Support has been added for editing the statistics information that is stored for a table or a partition. For example, you can run the following statement to set the number of rows for a table to 1000:

ALTER TABLE table1 UPDATE STATISTICS SET ('numRows'='1000');
        

For more information, see HIVE-12730 and the Apache wiki.

Support has been added for the SHOW CREATE DATABASE command, which prints out the DDL statement that was used to create a database:

SHOW CREATE DATABASE database1;
        

For more information, see HIVE-11706

Support has been added so that the IN clause in queries operates over multiple column references. The new syntax is boldfaced in the following example:

CREATE TABLE test (col1 int, col2 int);
INSERT INTO TABLE test VALUES (1, 1), (1, 2), (2, 1), (2, 2);
SELECT * FROM test;
+------------+------------+
| test.col1  | test.col2  |
+------------+------------+
| 1          | 1          |
| 1          | 2          |
| 2          | 1          |
| 2          | 2          |
+------------+------------+
SELECT * FROM test WHERE (col1, col2) IN ((1, 1));
+------------+------------+
| test.col1  | test.col2  |
+------------+------------+
| 1          | 1          |
+------------+------------+
        

For more information, see HIVE-11600

Support has been added for the following Hive UDFs:

All built-in Hive functions can be listed with the command SHOW FUNCTIONS; and a short description that explains what a function does is returned with the command DESCRIBE <function_name>; For more information about Hive functions, see the Apache wiki and Managing UDFs in the Cloudera enterprise documentation.

An optimization has been added to Hive on Spark that enables automatic caching of reused RDDs (Resilient Distributed Datasets). This optimization can improve query performance when the query or sub-query must scan a table multiple times. For example, TPC-DS query 39 is a query that requires multiple table scans. This optimization is disabled by default in CDH 6.0, but can be enabled by setting the hive.combine.equivalent.work.optimization property to true in the hive-site.xml file.

To configure this property in Cloudera Manager:

1. In the Admin Console, select the Hive service.
2. Click the Configuration tab.
3. Search for the HiveServer2 Advanced Configuration Snippet (Safety Valve) for hive-site.xml.

4. Enter the following property configuration information:

   • Name: hive.combine.equivalent.work.optimization
   • Value: true
   • Description: Enables dynamic RDD caching for HoS

   To disable this configuration, set the Value field to false.

   To set this configuration property in the XML editor, enter the following code:

   <property>
        <name>hive.combine.equivalent.work.optimization</name>
        <value>true</value>
        <description>Disables dynamic RDD caching for HoS</description>
   </property>
                  

5. Click Save Changes, and restart the service.

For more information see HIVE-10844 and HIVE-10550.

Support has been added for optimized hash tables for Hive on Spark to reduce memory overhead. This feature is enabled by default in CDH 6.0, but can be disabled by setting the hive.mapjoin.optimized.hashtable property to false in the hive-site.xml file. To configure this property in Cloudera Manager:

1. In the Admin Console, select the Hive service.
2. Click the Configuration tab.
3. Search for the HiveServer2 Advanced Configuration Snippet (Safety Valve) for hive-site.xml.

4. Enter the following property configuration information:

   • Name: hive.mapjoin.optimized.hashtable
   • Value: false
   • Description: Disables optimized hash tables for HoS

   To enable this configuration, set the Value field to true.

   To set this configuration property in the XML editor, enter the following code:

   <property>
        <name>hive.mapjoin.optimized.hashtable</name>
        <value>false</value>
        <description>Disables optimized hash tables for HoS</description>
   </property>
                  

5. Click Save Changes, and restart the service.

For more details, see HIVE-11182 and HIVE-6430.

• Listing available jobs in the metastore:

  sqoop job --list --meta-connect jdbc:oracle:thin:@//myhost:1521/ORCLCDB --meta-username ms_user --meta-password ms_password
  

• Creating a new job in the metastore:

  sqoop job --create myjob1 --meta-connect jdbc:oracle:thin:@//myhost:1521/ORCLCDB --meta-username ms_user --meta-password ms_password -- import --connect jdbc:mysql://mysqlhost:3306/sqoop --username sqoop --password sqoop --table "TestTable" -m 1
  

• Executing an existing job:

  sqoop job --exec myjob1 --meta-connect jdbc:oracle:thin:@//myhost:1521/ORCLCDB --meta-username ms_user --meta-password ms_password
  

• Showing the definition of an existing job:

  sqoop job --show myjob2 --meta-connect jdbc:oracle:thin:@//myhost:1521/ORCLCDB --meta-username ms_user --meta-password ms_password
  

• Deleting an existing job:

  sqoop job --delete myjob1 --meta-connect jdbc:oracle:thin:@//myhost:1521/ORCLCDB --meta-username ms_user --meta-password ms_password