ListBucketing
Skip to end of metadata
Go to start of metadata

Goal

The top level problem is as follows:

There are many tables of the following format:

  • create table T(a, b, c, ....., x) partitioned by (ds);

and the following queries need to performed efficiently:

  • select ... from T where x = 10;

The cardinality of 'x' is in 1000's per partition of T. Moreover, there is a skew for the values of 'x'. In general, there are ~10 values of 'x' which have a very large skew, and the remaining
values of 'x' have a small cardinality. Also, note that this mapping (values of 'x' with a high cardinality can change daily).

The above requirement can be solved in the following ways:

Basic Partitioning

Create a partition per value of 'x'.

  • create table T(a,b,c, .......) partitioned by (ds, x)
  • Advantages
    • Existing hive is good enough
  • Disadvantages
    • HDFS scalability: Number of files in HDFS increases.
    • HDFS scalability: Number of intermediate files in HDFS increases. For eg. if there are 1000 mappers and 1000 partitions, and each mapper gets atleast 1 row for each key, we will end up creating 1 million intermediate files.
    • Metastore Scalability: Will the metastore scale with the number of partitions.

List Bucketing

The basic idea here is as follows: Identify the keys with a high skew. Have one file per skewed key, and the remaining keys go into a separate file. This mapping is maintained in the metastore at a partition level, and is used by the
hive compiler to do input pruning. The list of skewed keys is stored at the table level (note that, this list can be initially supplied by the client periodically, and can be eventually updated when a new partition is being loaded).
For eg. the table maintains the list of skewed keys for 'x': 6, 20, 30, 40. When a new partition is being loaded, it will create 5 files (4 skewed keys + 1 file for all the remaining keys). The partition that got loaded will have the
following mapping: 6,20,30,40,others. This is similar to hash bucketing currently, where the bucket number determines the file number. Since the skewed keys need not be consecutive, the entire list of skewed keys need to be stored
in each partition.

When a query of the form

  • select .. from T where ds = '2012-04-15' and x = 30;

is issued, the hive compiler will only use the file corresponding to x=30 for the map-reduce job.

For a query of the form

  • select .. from T where ds = '2012-04-15' and x = 50;

the hive compiler will only use the file corresponding to x=others for the map-reduce job.

This approach is good under the following assumptions:

  • There are a few skewed keys per partition, which account for a significant percentage of the total data. In the above example, if the skewed keys (6,20,30 and 40) only occupy a small percentage of the data (say 20%), the queries of the form x=50 will still need to scan the remaining data (~80%).
  • The number of skewed keys are few. This list is stored in the metastore, so it does not make sense to store 1 million skewed keys per partition in the metastore.

This approach can be extended to the scenario when there are more than one clustered key. Say we want to optimize the queries of the form

  • select ... from T where x = 10 and y = 'b';
  • Extend the above approach. For each skewed value of (x,y), store the file offset. So, the metastore will have the mapping like: (10, 'a') -> 1, (10, 'b') -> 2, (20, 'c') -> 3, (others) -> 4.
    A query with all the clustering keys specified can be optimized easily. However, queries with some of the clustering keys specified:
    • select ... from T where x = 10;
    • select ... from T where y = 'b';

can only be used to prune very few files. It does not really matter, if the prefix of the clustering keys is specified or not. For eg. for x=10, the hive compiler can prune the file corresponding to (20, 'c').
And, for y='b', the files corresponding to (10, 'a') and (20, 'c') can be pruned. Hashing for others does not really help, when the complete key is not specified:

This approach does not scale in the following scenarios:

  • The number of skewed keys are very large. Creates a problem for metastore scalability.
  • In most of the cases, the number of clustered keys is more than one, and in the query, all the clustered keys are not specified.

Skewed Table vs List Bucketing Table

  • Skewed table is a table which has skewed information.
  • List Bucketing Table is a skewed table. In addition, it tells hive to use list bucketing feature on the skewed table: create sub-directories for skewed value.

Normal skewed table can be used for skewed join etc. You don't need to define it as list bucketing table if you don't use list bucketing feature.

List Bucketing Validation

Mainly due to sub-directory nature, list bucketing can't coexist with some features.

DDL

Compilation error will be thrown if list bucketing table coexists with

  • normal bucketing (clustered by, tablesample etc)
  • external table
  • "load data …"

DML

Compilation error will be thrown if list bucketing table coexists with

  • "insert into"
  • normal bucketing (clustered by, tablesample etc)
  • external table
  • non-RCfile due to merge.
  • non-partitioned table

Partitioning value should not be same as default list bucketing directory name.

Alter table Concatenate

Compilation error will be thrown if list bucketing table coexists with

  • non-RCfile
  • external table for alter table

Hive Enhancements

Hive needs to be extended to support the following:

Create Table

The table will be a skewed table. Skewed information will be created for all partitions.

For eg:

  • create table T (c1 string, c2 string) skewed by (c1) on ('x1');
  • create table T (c1 string, c2 string, c3 string) skewed by (c1, c2) on (('x1', 'x2'), ('y1', 'y2'));

'stored as DIRECTORIES' is optional parameter. It tells hive that if is not only skewed table but also list bucketing feature should apply: create sub-directories for skewed value.

Alter Table

The above is supported in table level only and not partition level.

It will

  • convert a table from a non-skewed table to a skewed table or
  • alter a skewed table's skewed column names and/or skewed values.

It won't

  • impact partitions created before the alter statement and
  • only impact partitions created afterwards.

The above will

  • turn off "skewed" feature from a table
  • make a table non-skewed
  • turn off "list bucketing" feature since a list bucketing table is a skewed table also

It won't

  • impact partitions created before the alter statement
  • only impact partitions created afterwards.

The above will

  • turn off "list bucketing"
  • doesn't turn off "skewed" feature from table since a "skewed" table can be a normal "skewed" table without list bucketing

The above will change list bucketing location map.

Design

When such a table is being loaded, it would be good to create a sub-directory per skewed key. The infrastructure similar to dynamic partitions can be used.
Alter table <T> partition <P> concatenate; needs to be changed to merge files per directory

Labels
  • No labels
  1. Does this feature require any changes to the metastore? If so can you please describe them? Thanks.

    1. Yes, it requires metastore change.

      We want to store the following information in metastore:
      1. skewed column names
      2. skewed column values
      3. mappings from skewed column value to directories.

      The above 3 will be added to MStorageDescriptor.java etc

  2. Please also describe any changes that will be made to public APIs including the following:

    • The metastore and/or HiveServer Thrift interfaces (note that this includes overloading functions that are already included in the current Thrift interfaces, as well as modifying or adding new Thrift structs/objects).
    • Hive Query Language, including new commands, extensions to existing commands, or changes to the output generated by commands (e.g. DESCRIBE FORMATTED TABLE).
    • New configuration properties.
    • Modifications to any of the public plugin APIs including SerDes and Hook/Listener interfaces,

    Also, if this feature requires any changes to the Metastore schema, those changes should be described in this document.

    Finally, please describe your plan for implementing this feature and getting it committed. Will it go in as a single patch or be split into several different patches.

    1. Yes, I will update document with any changes in the areas you mention.

      Here is plan:

      1. Implement End-to-end feature for single skewed column (DDL+DML) and go in as a single patch.
      2. Implement End-to-end feature for multiple skewed columns (DDL+DML) and go in as a single patch.
      3. Implement follow-ups and go in as a single patch.

      The #3 is a slot for those not critical but nice to have and not in #1 & #2 due to resource constraints etc.

  3. It wasn't clear to me from this wiki page what the benefit is of storing the skewed values "as directories" over just storing them as files as regular skew tables do? Tim, could you please elaborate on that?

    1. Different terms but refer to the same thing: create sub directory for skewed value and store record in file.

      Note that regular skew table doesn't create sub directory. It's different from non-skewed table because it has meta-data of skewed column name and values so that feature like skewed join can leverage it.

      Only list bucketing table creates sub directory for skewed-value. We use "stored as directories" to mark it.

      Hope it helps.

  4. Tim, thanks for responding but I am still missing something. I re-read the wiki page and here is my understanding. Please correct me if I am wrong.
    Let's take a hand-wavy example.
    Skewed table:
    create table t1 (x string) skewed by (error) on ('a', 'b') partitioned by dt location '/user/hive/warehouse/t1';
    will create the following files:
    /user/hive/warehouse/t1/dt=something/x=a.txt
    /user/hive/warehouse/t1/dt=something/x=b.txt
    /user/hive/warehouse/t1/dt=something/default

    List bucketing table:
    create table t2 (x string) skewed by (error) on ('a', 'b') partitioned by dt location '/user/hive/warehouse/t2' ;
    will create the following files:
    /user/hive/warehouse/t2/dt=something/x=a/data.txt
    /user/hive/warehouse/t2/dt=something/x=b/data.txt
    /user/hive/warehouse/t2/dt=something/default/data.txt

    Is that correct?

    In that case, why would a user ever choose to create sub-directories? Skewed joins would perform just well for regular skewed tables or list bucketing tables. Given that list bucketing introduces sub-directories it imposes restrictions on what other things users can and cannot do while regular skewed tables don't. So what would be someone's motivation to choose list bucketing over skewed tables?

    1. sorry for confusion. wiki requires polish to make it clear.

      I assume t2 has stored as directories.

      t1 doesn't have sub-directories but t2 has sub-directories. Directory structure looks like:
      /user/hive/warehouse/t1/dt=something/data.txt
      /user/hive/warehouse/t2/dt=something/x=a/data.txt
      /user/hive/warehouse/t2/dt=something/x=b/data.txt
      /user/hive/warehouse/t2/dt=something/default/data.txt

      "stored as directories" tells hive to create sub-directories.

      what's use case of t1? t1 can be used for skewed join since t1 has skewed column and value information.