E-MapReduce:Kudu connector

In the E-MapReduce (EMR) console, go to the Configure page of the Trino service. In the Service Configuration area, click the kudu.properties tab.

Examples:

1. Create a users table 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 required information, such as the primary key, column encoding and compression formats, and hash or range partitions.

2. View the table information.

   DESCRIBE kudu.default.users;

   The command returns output similar to the following:

      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. Insert data.

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

4. Query the data.

   SELECT * FROM kudu.default.users;

If you enable schema emulation in the connector's configuration file (etc/catalog/kudu.properties), tables are mapped to schemas based on a naming convention.

• If you set kudu.schema-emulation.enabled=true and kudu.schema-emulation.prefix=, tables are mapped as follows:

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

  Note

  Because Kudu does not natively support schemas, Presto creates a special table named $schemas to manage them.

• If you set kudu.schema-emulation.enabled=true and kudu.schema-emulation.prefix=presto::, tables are mapped as follows:

  Kudu table name	Presto table name
  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

  Because Kudu does not natively support schemas, Presto creates a special table named presto::$schemas to manage them.

The following table lists the data type mappings between Presto and Kudu.

To create a table, specify its columns, data types, and partition information. You can also specify other details, such as the column encoding format or compression format. The following is an example.

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
);

When creating a table, note the following:

• Primary key columns must be listed first. Partition key columns must be a subset of the primary key columns.

• number_of_replicas: The number of tablet replicas, which must be an odd number. If unspecified, the default from the Kudu Master configuration is used.

• Kudu supports hash and range partitions. A hash partition distributes rows into buckets based on a hash value. A range partition distributes rows based on an ordered range partition key, and its specific ranges must be created explicitly. Kudu supports multi-level partitioning. A table must have at least one hash or range partition. It can have multiple hash partitions but at most one range partition.

You can specify the following column properties in addition to the column name and data type.

The following is an example:

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 (...);

This section describes how to define hash and range partitions.

• Hash partition

  • Defining one partition group

    Use the partition_by_hash_columns and partition_by_hash_buckets table properties to specify the partition key columns and the number of partitions, respectively. All partition key columns must be a subset of the primary key columns. The following is an 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

    This example defines a hash partition on the col1 and col2 columns and distributes the data into four partitions.

  • Defining two partition groups

    To define two independent hash partition groups, you must also specify the partition_by_second_hash_columns and partition_by_second_hash_buckets table properties. The following is an 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

    This example defines two hash partition groups. The first group distributes rows into two partitions based on the col1 column. The second group distributes rows into three partitions based on the col2 column. The table has a total of six partitions (2 × 3).

• Range partition

  A Kudu table can have at most one range partition, which you can define using the partition_by_range_columns table property. When you create the table, you can define the partition ranges using the range_partitions table property. To manage range partitions on an existing table, use the kudu.system.add_range_partition and kudu.system.drop_range_partition stored procedures. The following is an 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

  This example defines a table with two hash partition groups and one range partition on the event_time column. The range is split at the timestamp 2018-01-01T00:00:00.

• Manage range partitions

  The following examples show partition operations:

  • Add a range partition

    CALL kudu.system.add_range_partition(<YOUR_SCHEMA_NAME>, <YOUR_TABLE_NAME>, <range_partition_as_json_string>)

  • Drop 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 where the table is located.
  <YOUR_TABLE_NAME>	The name of the table.
  <range_partition_as_json_string>	The upper and lower bounds of the range partition, specified as a JSON string. Use the format '{"lower": <value>, "upper": <value>}'. If the partition has multiple columns, use the format '{"lower": [<value_col1>,...], "upper": [<value_col1>,...]}'. The value format depends on the column's data type. The following list maps data types to their required 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: A Base64-encoded string. Note A null value indicates an unbounded boundary.

  The following is an example:

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

  Note

  This example adds a range partition to the events table in the myschema schema. The lower bound of the partition is 2018-01-01, which corresponds to the timestamp 2018-01-01T00:00:00.000. The upper bound is 2018-06-01.

  Use the SHOW CREATE TABLE statement to view existing range partitions. This information appears in the range_partitions table property.

Use the ALTER TABLE ... ADD COLUMN ... statement to add a column to an existing table. You can also specify column properties for the new column. For details about column properties, see Create a table.

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