Блог

Я хотел бы создать таблицу hive с помощью flink sql client.

Я мог бы успешно создать таблицу t2, но когда я запрашиваю t2, он жалуется

 Table options do not contain an option key 'connector' for discovering a connector.
 

Я установил тип выполнения как пакетный в conf/sql-client-defaults.файл yaml,

Я бы спросил, в чем здесь проблема. Спасибо!

 Flink SQL> use testdb1;

Flink SQL>  create table t2(id int,name string);
[INFO] Table has been created.

Flink SQL> select * from t2;
[ERROR] Could not execute SQL statement. Reason:
org.apache.flink.table.api.ValidationException: Table options do not contain an option key 'connector' for discovering a connector.
 

Проблема в том, что Flink не знает, где найти или поместить t2 — он должен быть связан с каким-либо источником или приемником данных, таким как файл, или раздел kafka, или база данных jdbc. Вам также необходимо указать формат, чтобы данные можно было сериализовать / десериализовать. Например:

 CREATE TABLE KafkaTable (
  `id` BIGINT,
  `name` STRING
) WITH (
  'connector' = 'kafka',
  'topic' = 'data',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'csv'
)
 

Дополнительные сведения см. в документации для конкретного используемого вами соединителя.

В конкретном случае Hive см. раздел Чтение и запись Hive. Здесь есть пример настройки таблицы для записи в Hive, которая выглядит примерно так:

 SET table.sql-dialect=hive;
CREATE TABLE hive_table (
  id BIGINT,
  name STRING
) PARTITIONED BY (dt STRING, hr STRING) STORED AS parquet TBLPROPERTIES (
  'partition.time-extractor.timestamp-pattern'='$dt $hr:00:00',
  'sink.partition-commit.trigger'='partition-time',
  'sink.partition-commit.delay'='1 h',
  'sink.partition-commit.policy.kind'='metastore,success-file'
);