E-MapReduce:Kudu connector

Log on to the E-MapReduce (EMR) console and go to the Configure tab of the Trino service page. On the Configure tab, click kudu.properties. Modify or add the configuration items that are described in the following table based on your business requirements.

Examples:

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 primary key, the encoding format or compression format of 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. Insert data into the table.

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

4. Query data from the table.

   SELECT * FROM kudu.default.users;

If you enable the schema emulation feature in the kudu.properties configuration file for the Kudu connector in the etc/catalog/ directory, Kudu tables are mapped to schemas based on the naming conventions.

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

  Kudu table name	Presto table name
  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 you configure kudu.schema-emulation.enabled=true and kudu.schema-emulation.prefix=presto::, the mappings listed in the following table prevail.

  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

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

The following table describes the mappings between Presto data types and Kudu data types.

When you create a table, you must specify columns, data types, and partition information. You can also specify the column encoding format or compression format based on your business requirements. 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
);

Take note of the following items when you configure parameters for the CREATE TABLE statement:

• 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. 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.

In addition to column names and data types, you can also specify other column properties.

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

A table can contain only one range partition or at least one hash partition.

• 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 field, 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 drop 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>)

  • 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 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 configure this parameter in the '{"lower": <value>, "upper": <value>}' JSON format. If the partition has multiple columns, you must configure 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 returned results, the table property range_partitions indicates the partition information of the table.

You can execute the ALTER TABLE ... ADD COLUMN ... statement to add a column to an existing table. You can also use column properties to add columns. For more information about the column properties, see Create a table.

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