E-MapReduce:Kudu connector

1. Create a table named users in the default schema.

   CREATE TABLE kudu.default.users (
     user_id int WITH (primary_key = true),
     first_name varchar,
     last_name varchar
   ) WITH (
     partition_by_hash_columns = ARRAY['user_id'],
     partition_by_hash_buckets = 2
   );

   Note When you create a table, you must specify the required table information, such as the encoding format or compression format of primary keys and columns, and the hash partitions or range partitions.
2. View information about the table.

   DESCRIBE kudu.default.users;

   Information similar to the following output is returned:

      Column   |  Type   |                      Extra                      | Comment
   ------------+---------+-------------------------------------------------+---------
    user_id    | integer | primary_key, encoding=auto, compression=default |
    first_name | varchar | nullable, encoding=auto, compression=default    |
    last_name  | varchar | nullable, encoding=auto, compression=default    |
   (3 rows)

3. Inserts data into the table.

   INSERT INTO kudu.default.users VALUES (1, 'Donald', 'Duck'), (2, 'Mickey', 'Mouse');

4. Query data of the table.

   SELECT * FROM kudu.default.users;

• If kudu.schema-emulation.enabled=true and kudu.schema-emulation.prefix= are configured, the mappings listed in the following table prevail.

  Kudu table name	Table name used in Presto
  orders	kudu.default.orders
  part1.part2	kudu.part1.part2
  x.y.z	kudu.x."y.z"

  Note Kudu does not support schemas. Presto creates a special table named $schemas to manage schemas.
• If kudu.schema-emulation.enabled=true and kudu.schema-emulation.prefix=presto:: are configured, the mappings listed in the following table prevail.

  Kudu table name	Table name used in Presto
  orders	kudu.default.orders
  part1.part2	kudu.default."part1.part2"
  x.y.z	kudu.default."x.y.z"
  presto::part1.part2	kudu.part1.part2
  presto::x.y.z	kudu.x."y.z"

  Note Kudu does not support schemas. Presto creates a special table named presto::$schemas to manage schemas.

CREATE TABLE user_events (
  user_id int WITH (primary_key = true),
  event_name varchar WITH (primary_key = true),
  message varchar,
  details varchar WITH (nullable = true, encoding = 'plain')
) WITH (
  partition_by_hash_columns = ARRAY['user_id'],
  partition_by_hash_buckets = 5,
  number_of_replicas = 3
);

• Primary key columns must be specified before other columns, and only a primary key column can be configured as a partition key column.
• The number_of_replicas parameter is optional. This parameter specifies the number of tablet replicas and must be set to an odd number. If you do not configure this parameter, the default replication factor from the Kudu master configuration is used.
• Kudu supports hash partitions and range partitions. A hash partition distributes rows by hash value to one of many buckets. A range partition distributes rows by using an ordered range partition key. The concrete range partitions must be explicitly created. Kudu supports multi-level partitioning. A table must contain at least one hash or range partition. A table can contain only one range partition but multiple hash partitions.

CREATE TABLE mytable (
  name varchar WITH (primary_key = true, encoding = 'dictionary', compression = 'snappy'),
  index bigint WITH (nullable = true, encoding = 'runlength', compression = 'lz4'),
  comment varchar WITH (nullable = true, encoding = 'plain', compression = 'default'),
   ...
) WITH (...);

• Define hash partitions
  • Define one partition group
    You can use the table property partition_by_hash_columns to specify partition key columns and use the table property partition_by_hash_buckets to specify the number of partitions. The partition key columns must be a subset of primary key columns. Example:

    CREATE TABLE mytable (
      col1 varchar WITH (primary_key=true),
      col2 varchar WITH (primary_key=true),
      ...
    ) WITH (
      partition_by_hash_columns = ARRAY['col1', 'col2'],
      partition_by_hash_buckets = 4
    )

    Note In this example, col1 and col2 columns are defined as hash partition key columns, and data is distributed to four partitions.
  • Define two partition groups
    If you want to define two independent hash partition groups, in addition to the table properties specified in the preceding example, you must specify the table properties partition_by_second_hash_columns and partition_by_second_hash_buckets. Example:

    CREATE TABLE mytable (
      col1 varchar WITH (primary_key=true),
      col2 varchar WITH (primary_key=true),
      ...
    ) WITH (
      partition_by_hash_columns = ARRAY['col1'],
      partition_by_hash_buckets = 2,
      partition_by_second_hash_columns = ARRAY['col2'],
      partition_by_second_hash_buckets = 3
    )

    Note In this example, two hash partition groups are defined. In the first hash partition group, rows are distributed to two partitions based on the col1 column. In the second hash partition group, rows are distributed to three partitions based on the col2 column. In this case, the total number of partitions in the table is 6 (2 × 3).
• Define range partitions
  A Kudu table can contain only one range partition, which can be defined by using the table property partition_by_range_columns. When you create a table, you can use the table property range_partitions to define the range of the partition. You can use the table properties kudu.system.add_range_partition and kudu.system.drop_range_partition to manage the range partitions of existing tables. Example:

  CREATE TABLE events (
    rack varchar WITH (primary_key=true),
    machine varchar WITH (primary_key=true),
    event_time timestamp WITH (primary_key=true),
    ...
  ) WITH (
    partition_by_hash_columns = ARRAY['rack'],
    partition_by_hash_buckets = 2,
    partition_by_second_hash_columns = ARRAY['machine'],
    partition_by_second_hash_buckets = 3,
    partition_by_range_columns = ARRAY['event_time'],
    range_partitions = '[{"lower": null, "upper": "2018-01-01T00:00:00"},
                         {"lower": "2018-01-01T00:00:00", "upper": null}]'
  )

  Note In this example, two hash partition groups and one range partition are defined. The table is range partitioned on the event_time column, and data is split based on 2018-01-01T00:00:00.
• Manage range partitions

  You can use a stored procedure to add a range partition to or remove a range partition from an existing table.

  Examples:

  • Add a range partition:

    CALL kudu.system.add_range_partition(<your_schema_name>, <your_table_name>, <range_partition_as_json_string>)

  • Remove a range partition:

    CALL kudu.system.drop_range_partition(<your_schema_name>, <your_table_name>, <range_partition_as_json_string>)

  Parameter	Description
  <your_schema_name>	The schema to which the table belongs.
  <your_table_name>	The name of the table.
  <range_partition_as_json_string>	The upper and lower bounds of the range partition. You must set this parameter in the '{"lower": <value>, "upper": <value>}' JSON format. If the partition has multiple columns, you must set this parameter in the '{"lower": [<value_col1>,...], "upper": [<value_col1>,...]}' format. The specific value formats of the upper and lower bounds depend on the data types of columns. Mappings between data types and JSON string formats: BIGINT: '{"lower": 0, "upper": 1000000}' SMALLINT: '{"lower": 10, "upper": null}' VARCHAR: '{"lower": "A", "upper": "M"}' TIMESTAMP: '{"lower": "2018-02-01T00:00:00.000", "upper": "2018-02-01T12:00:00.000"}' BOOLEAN: '{"lower": false, "upper": true}' VARBINARY: Base64-encoded strings Note If you set this parameter to null, the partition is unbounded.

  Example:

  CALL kudu.system.add_range_partition('myschema', 'events', '{"lower": "2018-01-01", "upper": "2018-06-01"}')

  Note In this example, a range partition is added to the events table in the myschema schema. The lower bound of the partition is 2018-01-01, whose accurate value is 2018-01-01T00:00:00.000. The upper bound of the partition is 2018-06-01.

  You can execute the SHOW CREATE TABLE statement to query the existing range partition of the table. In the return results, the table property range_partitions indicates the partition information of the table.

ALTER TABLE mytable ADD COLUMN extraInfo varchar WITH (nullable = true, encoding = 'plain')