Ключ при создании KSQL Stream - PullRequest
Новая
       11

Ключ при создании KSQL Stream

0 голосов
/ 22 июня 2019

1) Требуется ли ключ в потоке, где вы хотите выполнить функцию агрегирования.Я прочитал несколько блогов, а также рекомендацию от Confluent, что для работы функции агрегации требуется KEY 

CREATE STREAM Employee    (EmpId BIGINT,     EmpName VARCHAR,    
DeptId BIGINT,    SAL BIGINT)   WITH (KAFKA_TOPIC='EmpTopic', 
   VALUE_FORMAT='JSON');

При определении выше Stream я не определил ни одного KEY (ROWKEY равен NULL).Основная тема «EmpTopic» также не является KEY.

Я выполняю функцию агрегирования в потоке. 

CREATE TABLE SALBYDEPT AS
    SELECT DeptId, 
             SUM(SAL) 
      FROM Employee 
      GROUP BY DeptId;

Пожалуйста, подтвердите, требует ли выполнение функции агрегации в вышеупомянутом потоке KEY on 'Поток сотрудника, т.е. NOT NULL ROWKEY в потоке «Сотрудник»

2) Согласно документации Confluent, «Управление окнами» позволяет вам контролировать, как группировать записи с одинаковым ключом для операций с состоянием, таких как агрегации или объединения, во времяKSQL отслеживает окна для каждого ключа записи ".Пожалуйста, помогите мне понять значение вышеприведенного заявления.Требуется ли, чтобы поток имел ключ NOT NULL?

3) Будет ли JOIN на Stream-Table сохранять ключ 

CREATE TABLE users 
  (registertime BIGINT, 
   userid VARCHAR, 
   gender VARCHAR, 
   regionid VARCHAR) 
  WITH (KAFKA_TOPIC = 'users', 
        VALUE_FORMAT='JSON', 
        KEY = 'userid');

CREATE STREAM pageviews 
  (viewtime BIGINT, 
   userid VARCHAR, 
   pageid VARCHAR) 
  WITH (KAFKA_TOPIC='pageviews', 
        VALUE_FORMAT='DELIMITED', 
        KEY='pageid', 
        TIMESTAMP='viewtime');

CREATE STREAM pageviews_transformed as 
  SELECT viewtime, 
         userid, 
         pageid, 
         TIMESTAMPTOSTRING(viewtime, 'yyyy-MM-dd HH:mm:ss.SSS') AS timestring 
  FROM pageviews 

CREATE STREAM pageviews_enriched AS 
  SELECT pv.viewtime, 
         pv.userid AS userid, 
         pv.pageid, 
         pv.timestring, 
         u.gender, 
         u.regionid, 
         u.interests, 
         u.contactinfo 
  FROM pageviews_transformed pv 
  LEFT JOIN users u ON pv.userid = u.userid;

Will JOIN on Stream-Table retain the 'UserId' as ROWKEY in the new Stream 'pageviews_enriched'

4) Я видел несколько примеров из Confluent на Github, где Streamиспользуемый в JOIN не является KEY'ed.Но согласно документации, Stream должен иметь NOT NULL ROWKEY, участвующий в JOIN.Подтвердите, что в потоке указано NOT NULL ROWKEY.

Соединение Stream-Stream и Соединение Stream-Table.В приведенном ниже примере я выполняю JOIN on Stream с NULL ROWKEY и таблицей.Это действительно? 

CREATE TABLE users 
  (registertime BIGINT, 
   userid VARCHAR, 
   gender VARCHAR, 
   regionid VARCHAR) 
  WITH (KAFKA_TOPIC = 'users', 
        VALUE_FORMAT='JSON', 
        KEY = 'userid');

CREATE STREAM pageviews 
  (viewtime BIGINT, 
   userid VARCHAR, 
   pageid VARCHAR) 
  WITH (KAFKA_TOPIC='pageviews', 
        VALUE_FORMAT='DELIMITED', 
        TIMESTAMP='viewtime');

CREATE STREAM pageviews_transformed as 
  SELECT viewtime, 
         userid, 
         pageid, 
         TIMESTAMPTOSTRING(viewtime, 'yyyy-MM-dd HH:mm:ss.SSS') AS timestring 
  FROM pageviews 

CREATE STREAM pageviews_enriched AS 
  SELECT pv.viewtime, 
         pv.userid AS userid, 
         pv.pageid, 
         pv.timestring, 
         u.gender, 
         u.regionid, 
         u.interests, 
         u.contactinfo 
  FROM pageviews_transformed pv 
  LEFT JOIN users u ON pv.userid = u.userid;

1 Ответ

1 голос
/ 23 июня 2019 
CREATE TABLE SALBYDEPT AS
   SELECT DeptId, 
            SUM(SAL) 
     FROM Employee 
     GROUP BY DeptId;
  1. Пожалуйста, подтвердите, требует ли выполнение функции агрегации в вышеупомянутом потоке КЛЮЧ на потоке «Сотрудник», т.е. НЕ НУЛЕВОЙ ROWKEY на потоке «Сотрудник»

Вам не нужен ключ в этом потоке.Ключ созданной таблицы будет DeptId.

Согласно документации Confluent, «Оконное управление позволяет вам контролировать, как группировать записи, имеющие один и тот же ключ для операций с состоянием, таких как агрегации или объединения, в промежутки времени. KSQL отслеживает окна для каждого ключа записи».Пожалуйста, помогите мне понять значение вышеприведенного заявления.Требуется ли, чтобы у потока был ключ NOT NULL?

Это означает, что при создании агрегации вы можете делать это в течение временного окна, и это временное окно является частью ключа сообщения.,Например, вместо агрегирования all employee SAL (sales?), Вы можете сделать это за промежуток времени, возможно, каждый час или день.В этом случае у вас будет агрегатный ключ (DeptId) в сочетании с клавишей window (например, для почасовых 2019-06-23 06:00:00, 2019-06-23 07:00:00, 2019-06-23 08:00:00 и т. Д.)

Будет ли JOIN на Stream-Table сохранять KEY

Он сохранит ключ потока, если вы не включите PARTITION BY в DDL.

Я видел несколько примеров из Confluent на Github, где Stream, используемый в JOIN, не KEY'ed.Но согласно документации, Stream должен иметь NOT NULL ROWKEY, участвующий в JOIN.Подтвердите, что в потоке NOT NULL ROWKEY.

У вас есть ссылка на конкретную документацию, на которую вы ссылаетесь?В то время как таблица действительно должна иметь ключ, stream не требует (KSQL может справиться с этим под прикрытием; я не уверен).

...