Partitioned Tables

Users are now able to create and cache partitioned tables. Different from RDD partitions that correspond to Hadoop splits, Hive "partitions" are analogous to indexes. Each partition is represented by an RDD and identifiable by the set of runtimes values for virtual partitioning columns that specified at table creation.

In-memory partitioned tables also adhere to partition-level cache policies, which can be toggled through the shark.cache.policy table property and customized by implementing the CachePolicy interface (an LRU implementation is provided).

During query execution, Shark uses partitioning keys to automatically filter input partitions. This feature can is be combined with RDD-partition level pruning on non-partitioned columns to further decrease the amount of data that needs to be fetched and scanned.

Tachyon Support

The complete set of commands supported for in-memory Shark tables stored in the Spark-managed heap are now supported for Tachyon-backed tables as well. This includes Hive-partitioned tables and table recovery features added in this 0.8.1 release.

Spark Integration

Stability and usability improvements have been added to reduce friction in converting between native Spark RDDs and Shark tables. A key pair of features are SharkContext’s sqlRdd() functions and rddToTable implicit conversions, both of which can automatically deduce data types and update necessary metadata for transitions between RDDs and Shark tables. Both can be tested by launching a Shark shell (SHARK_HOME/bin/shark-shell).

Table Generating Functions (TGFs)

Shark can now call into libraries that generate tables through TGFs. This enables Shark to easily access external libraries, such as Spark’s machine learning library (MLLIB).

Calls can be made into TGFs by executing GENERATE tgf_name(params) or GENERATE tgf_name(params) SAVE table_name. TGFs are flexible and can take arbitrary tables and parameters as inputs and produce a new table with an accompanying schema.

Other improvements

• To reduce the overhead for Hive-partitioned table scans, job configurations are only broadcasted once and shared throughout the entire read operation over a partitioned table. Previously, these configuration variables were broadcasted once per partition.
• Commands that use COUNT DISTINCT operations, but don’t include grouping keys, are automatically rewritten to generate query plans that can take advantage of multiple reducers (set through the mapred.reduce.tasks property) and increased parallelism. This eliminates the previous single-reducer bottleneck.

Credits

Michael Armbrust - test util improvements Harvey Feng - Tachyon support, caching semantics, partitioned table, release manager Ali Ghodsi - table generating functions Mark Hamstra - build fix Cheng Hao - work on removing Hive operator dependencies Nandu Jayakumar - code and style cleanup Andy Konwinski - build script fix Haoyuan Li - Tachyon integration Xi Liu - byte buffer overflow bug fix Sundeep Narravula - support for database namespaces for cached tables, code cleanup Patrick Wendell - bug fix Reynold Xin - caching semantics, Spark integration, miscellaneous bug fixes

Thanks to everyone who contributed!