E-MapReduce:Conector Kudu

connector.name=kudu

## Required: Kudu master address(es). Separate multiple addresses with commas.
## Supported formats: example.com, example.com:7051, 192.0.2.1, 192.0.2.1:7051,
##                    [2001:db8::1], [2001:db8::1]:7051, 2001:db8::1
## Change localhost to the IP address or hostname of your Kudu master node (e.g., master-1-1).
kudu.client.master-addresses=localhost

## Schema emulation lets Trino map Kudu tables to schemas using naming conventions.
## By default, all tables appear in the "default" schema.
#kudu.schema-emulation.enabled=false

## Prefix used when schema emulation is enabled. Standard prefix is "presto::". Empty prefix is also valid.
## Required only when kudu.schema-emulation.enabled=true.
#kudu.schema-emulation.prefix=

## Advanced Kudu Java client configuration

## Timeout for administrative operations (e.g., CREATE TABLE, DROP TABLE). Default: 30s.
#kudu.client.default-admin-operation-timeout=30s

## Timeout for user operations. Default: 30s.
#kudu.client.default-operation-timeout=30s

## Timeout for waiting on data from a socket. Default: 10s.
#kudu.client.default-socket-read-timeout=10s

## Whether to disable Kudu client statistics collection. Default: false.
#kudu.client.disable-statistics=false

Emulação de esquema desativada (padrão)

Com a emulação de esquema desativada, todas as tabelas do Kudu aparecem no esquema default.

Consulte a tabela orders usando seu nome totalmente qualificado:

SELECT * FROM kudu.default.orders;

Se você definir kudu como catálogo e default como esquema, a consulta será simplificada para:

SELECT * FROM orders;

Caso o nome da tabela contenha caracteres especiais, envolva-o entre aspas duplas:

SELECT * FROM kudu.default."special.table!";

Exemplo rápido: criar e consultar uma tabela

1. Crie uma tabela chamada users no esquema default:

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

   Ao criar uma tabela, especifique a chave primária, a codificação da coluna ou o formato de compactação e as informações de partição (hash ou intervalo).

2. Inspecione o esquema da tabela:

   DESCRIBE kudu.default.users;

   Saída esperada:

      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. Insira linhas:

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

4. Consulte os dados:

   SELECT * FROM kudu.default.users;

Emulação de esquema ativada

Ative a emulação de esquema definindo kudu.schema-emulation.enabled=true no arquivo etc/catalog/kudu.properties. As tabelas do Kudu serão então mapeadas para esquemas do Trino com base em seus nomes.

Mapeamento com prefixo vazio (kudu.schema-emulation.prefix=)

O Kudu não oferece suporte a esquemas. O Trino cria uma tabela especial chamada $schemas para gerenciá-los.

Mapeamento com o prefixo padrão (kudu.schema-emulation.prefix=presto::)

Com o prefixo padrão, o Trino cria uma tabela especial chamada presto::$schemas para gerenciar os esquemas.

De Trino para Kudu

Ao gravar dados do Trino para o Kudu, aplicam-se os seguintes mapeamentos de tipos:

Os seguintes tipos do Trino não são suportados: TIME, JSON, TIME WITH TIMEZONE, TIMESTAMP WITH TIME ZONE, INTERVAL YEAR TO MONTH, INTERVAL DAY TO SECOND, ARRAY, MAP, IPADDRESS.

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

Propriedades da coluna

Especifique as propriedades da coluna na cláusula WITH:

Exemplo com codificação e compactação explícitas:

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

Design de partição

O Kudu suporta partições hash e partições por intervalo. Uma tabela deve ter pelo menos uma partição de qualquer um desses tipos.

Definir partições hash

Especifique partition_by_hash_columns (colunas de chave de partição) e partition_by_hash_buckets (número de buckets). As colunas de chave de partição devem ser um subconjunto das colunas de chave primária.

Um grupo de partição hash:

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

col1 e col2 são as colunas de chave de partição hash. As linhas são distribuídas em 4 buckets.

Dois grupos independentes de partição hash:

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

O primeiro grupo distribui as linhas por col1 em 2 buckets; o segundo distribui por col2 em 3 buckets. O número total de partições é 6 (2 x 3).

Definir partições por intervalo

Use partition_by_range_columns para especificar as colunas de partição por intervalo e range_partitions para definir os limites iniciais da partição.

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}]'
);

Esta tabela possui dois grupos de partição hash e uma partição por intervalo em event_time, dividida em 2018-01-01T00:00:00.

Gerenciar partições por intervalo

Adicione ou remova partições por intervalo em uma tabela existente usando procedimentos armazenados:

-- Add a range partition
CALL kudu.system.add_range_partition(<schema_name>, <table_name>, <range_partition_as_json_string>)

-- Drop a range partition
CALL kudu.system.drop_range_partition(<schema_name>, <table_name>, <range_partition_as_json_string>)

Parâmetros:

Formatos de valor JSON por tipo de dado:

Exemplo: Adicionar uma partição por intervalo à tabela events no esquema myschema, abrangendo registros de 2018-01-01 a 2018-06-01:

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

O limite inferior "2018-01-01" é interpretado como 2018-01-01T00:00:00.000.

Para visualizar todas as partições por intervalo existentes em uma tabela, execute SHOW CREATE TABLE. A propriedade range_partitions na saída lista os limites atuais da partição.

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