ksql, выберите на столе ничего не показывать - PullRequest
Новая
       15

ksql, выберите на столе ничего не показывать

0 голосов
/ 07 июня 2018

Я создал исходную подписку на тему, у которой есть входное сообщение, подобное этому: 

{
  "ip_router": "",
  "ip_lan": "",
  "isdn": "2046573688",
  "end_datetime": "",
  "shop_code": "1000405100",
  "reg_type_id": "5131615",
  "contract_id": "",
  "update_datetime": "20170801171355",
  "project": "",
  "telecom_service_id": "2",
  "local_speed": "",
  "password": "",
  "price_plan": "",
  "vip": "",
  "local_price_plan": "",
  "sub_id": "1083168000",
  "sta_datetime": "20090511152847",
  "update_number_1": "1",
  "act_status": "000",
  "network_class": "",
  "limit_usage": "",
  "num_reset_zone": "",
  "deposit": "",
  "create_user": "TUDV_POPBGG",
  "num_of_computer": "",
  "cust_id": "10922428129",
  "status": "2",
  "active_datetime": "20090511152102",
  "ip_view": "",
  "channel_type_id": "",
  "ip_wan": "",
  "imsi": "452049760887694",
  "infrastructure_type": "",
  "product_code": "HPN03",
  "expire_datetime": "",
  "speed": "",
  "private_ip": "",
  "update_user": "MIGRATE",
  "ip_static": "",
  "vlan": "",
  "sub_type": "",
  "create_datetime": "20090511152102",
  "is_info_completed": "1",
  "pay_type": "2",
  "up_link": "",
  "promotion_code": "",
  "technology": "",
  "offer_id": "400001035",
  "dev_staff_id": "",
  "account_id": "",
  "deploy_accept_date": "",
  "serial": "8984049767000887694",
  "group_id": "",
  "ip_gateway": "",
  "first_connect": "",
  "org_product_code": "MIGRATE",
  "start_money": "100000",
  "keep_alive": "",
  "account": ""
}

А затем я создал поток и таблицу на нем: 

CREATE STREAM str_subscriber_json (sub_id VARCHAR,contract_id VARCHAR,cust_id VARCHAR,account_id VARCHAR,telecom_service_id VARCHAR,isdn VARCHAR,imsi VARCHAR,serial VARCHAR,status VARCHAR,product_code VARCHAR,offer_id VARCHAR,act_status VARCHAR,sta_datetime BIGINT,active_datetime BIGINT,sub_type VARCHAR,end_datetime BIGINT,expire_datetime BIGINT,shop_code VARCHAR,dev_staff_id VARCHAR,promotion_code VARCHAR,vip VARCHAR,account VARCHAR,create_datetime BIGINT,create_user VARCHAR,update_datetime BIGINT,update_user VARCHAR,deposit VARCHAR,limit_usage VARCHAR,password VARCHAR,org_product_code VARCHAR,num_reset_zone VARCHAR,start_money VARCHAR,is_info_completed VARCHAR,channel_type_id VARCHAR,first_connect VARCHAR,speed VARCHAR,keep_alive VARCHAR,price_plan VARCHAR,local_price_plan VARCHAR,project VARCHAR,local_speed VARCHAR,technology VARCHAR,network_class VARCHAR,infrastructure_type VARCHAR,deploy_accept_date BIGINT,group_id VARCHAR,ip_static VARCHAR,ip_view VARCHAR,ip_lan VARCHAR,ip_wan VARCHAR,ip_gateway VARCHAR,ip_router VARCHAR,num_of_computer VARCHAR,vlan VARCHAR,up_link VARCHAR,private_ip VARCHAR,pay_type VARCHAR,update_number_1 VARCHAR,reg_type_id VARCHAR,type VARCHAR,update_number VARCHAR) WITH (KAFKA_TOPIC='subscriber', VALUE_FORMAT='JSON');

CREATE TABLE tbl_subscriber_json (sub_id VARCHAR,contract_id VARCHAR,cust_id VARCHAR,account_id VARCHAR,telecom_service_id VARCHAR,isdn VARCHAR,imsi VARCHAR,serial VARCHAR,status VARCHAR,product_code VARCHAR,offer_id VARCHAR,act_status VARCHAR,sta_datetime BIGINT,active_datetime BIGINT,sub_type VARCHAR,end_datetime BIGINT,expire_datetime BIGINT,shop_code VARCHAR,dev_staff_id VARCHAR,promotion_code VARCHAR,vip VARCHAR,account VARCHAR,create_datetime BIGINT,create_user VARCHAR,update_datetime BIGINT,update_user VARCHAR,deposit VARCHAR,limit_usage VARCHAR,password VARCHAR,org_product_code VARCHAR,num_reset_zone VARCHAR,start_money VARCHAR,is_info_completed VARCHAR,channel_type_id VARCHAR,first_connect VARCHAR,speed VARCHAR,keep_alive VARCHAR,price_plan VARCHAR,local_price_plan VARCHAR,project VARCHAR,local_speed VARCHAR,technology VARCHAR,network_class VARCHAR,infrastructure_type VARCHAR,deploy_accept_date BIGINT,group_id VARCHAR,ip_static VARCHAR,ip_view VARCHAR,ip_lan VARCHAR,ip_wan VARCHAR,ip_gateway VARCHAR,ip_router VARCHAR,num_of_computer VARCHAR,vlan VARCHAR,up_link VARCHAR,private_ip VARCHAR,pay_type VARCHAR,update_number_1 VARCHAR,reg_type_id VARCHAR,type VARCHAR,update_number VARCHAR) WITH (KAFKA_TOPIC='subscriber', VALUE_FORMAT='JSON', KEY = 'sub_id' );

Я пытался использовать ksqlдля теста: 

SELECT * FROM str_subscriber_json;

(результат печати, когда я добавляю новый json в тему подписчика) 

SELECT * FROM tbl_subscriber_json;

(ничего не показывалось, когдаЯ ставлю новый JSON в тему подписчика)

Поэтому, пожалуйста, уточните мне, что не так в этом случае?

Большое вам спасибо.

1 Ответ

0 голосов
/ 07 июня 2018

Сводка

Ваши сообщения должны быть набраны ключом.Если у вас нет ключа сообщения, то семантика TABLE не имеет никакого смысла (поскольку вы не можете показать значение для ключа, если ключа нет).

Я воспроизвел ваш пример, используя kafkacat для создания сообщений с ключом и без него.

Тест 1 - без ключа

Создание тестового сообщения 

$ echo '{ "ip_router": "", "ip_lan": "", "isdn": "2046573688", "end_datetime": "", "shop_code": "1000405100", "reg_type_id": "5131615", "contract_id": "", "update_datetime": "20170801171355", "project": "", "telecom_service_id": "2", "local_speed": "", "password": "", "price_plan": "", "vip": "", "local_price_plan": "", "sub_id": "1083168000", "sta_datetime": "20090511152847", "update_number_1": "1", "act_status": "000", "network_class": "", "limit_usage": "", "num_reset_zone": "", "deposit": "", "create_user": "TUDV_POPBGG", "num_of_computer": "", "cust_id": "10922428129", "status": "2", "active_datetime": "20090511152102", "ip_view": "", "channel_type_id": "", "ip_wan": "", "imsi": "452049760887694", "infrastructure_type": "", "product_code": "HPN03", "expire_datetime": "", "speed": "", "private_ip": "", "update_user": "MIGRATE", "ip_static": "", "vlan": "", "sub_type": "", "create_datetime": "20090511152102", "is_info_completed": "1", "pay_type": "2", "up_link": "", "promotion_code": "", "technology": "", "offer_id": "400001035", "dev_staff_id": "", "account_id": "", "deploy_accept_date": "", "serial": "8984049767000887694", "group_id": "", "ip_gateway": "", "first_connect": "", "org_product_code": "MIGRATE", "start_money": "100000", "keep_alive": "", "account": "" }' \
| kafkacat -b localhost:9092 -P -t subscriber

Потоковый вывод

Обратите внимание на null во втором столбце - этоключ (первый столбец - это отметка времени сообщения; остальные столбцы - это объявленные поля в сообщении) 

ksql> select * from str_subscriber_json;
1528368689380 | null | 1083168000 |  | 10922428129 |  | 2 | 2046573688 | 452049760887694 | 8984049767000887694 | 2 | HPN03 | 400001035 | 000 | 20090511152847 | 20090511152102 |  | 0 | 0 | 1000405100 |  |  |  |  | 20090511152102 | TUDV_POPBGG | 20170801171355 | MIGRATE |  |  |  | MIGRATE |  | 100000 | 1 |  |  |  |  |  |  |  |  |  |  |  | 0 |  |  |  |  |  |  |  |  |  |  |  | 2 | 1 | 5131615 | null | null

Вывод таблицы 

ksql> SELECT * FROM tbl_subscriber_json;

(без вывода)

Тест 2 - с установленным ключом

Здесь ключ произвольно установлен на 1, используя флаг kafkacat '-K, чтобы указать : в качестве разделителя ключ / значение,

$ echo '1:{ "ip_router": "", "ip_lan": "", "isdn": "2046573688", "end_datetime": "", "shop_code": "1000405100", "reg_type_id": "5131615", "contract_id": "", "update_datetime": "20170801171355", "project": "", "telecom_service_id": "2", "local_speed": "", "password": "", "price_plan": "", "vip": "", "local_price_plan": "", "sub_id": "1083168000", "sta_datetime": "20090511152847", "update_number_1": "1", "act_status": "000", "network_class": "", "limit_usage": "", "num_reset_zone": "", "deposit": "", "create_user": "TUDV_POPBGG", "num_of_computer": "", "cust_id": "10922428129", "status": "2", "active_datetime": "20090511152102", "ip_view": "", "channel_type_id": "", "ip_wan": "", "imsi": "452049760887694", "infrastructure_type": "", "product_code": "HPN03", "expire_datetime": "", "speed": "", "private_ip": "", "update_user": "MIGRATE", "ip_static": "", "vlan": "", "sub_type": "", "create_datetime": "20090511152102", "is_info_completed": "1", "pay_type": "2", "up_link": "", "promotion_code": "", "technology": "", "offer_id": "400001035", "dev_staff_id": "", "account_id": "", "deploy_accept_date": "", "serial": "8984049767000887694", "group_id": "", "ip_gateway": "", "first_connect": "", "org_product_code": "MIGRATE", "start_money": "100000", "keep_alive": "", "account": "" }' \
| kafkacat -b localhost:9092 -P -t subscriber -K:

Потоковый вывод

Обратите внимание на 1 во втором столбце - это ключ (первый столбец - это отметка времени сообщения; остальные столбцы - объявленныеполя в сообщении) 

1528368781916 | 1 | 1083168000 |  | 10922428129 |  | 2 | 2046573688 | 452049760887694 | 8984049767000887694 | 2 | HPN03 | 400001035 | 000 | 20090511152847 | 20090511152102 |  | 0 | 0 | 1000405100 |  |  |  |  | 20090511152102 | TUDV_POPBGG | 20170801171355 | MIGRATE |  |  |  | MIGRATE |  | 100000 | 1 |  |  |  |  |  |  |  |  |  |  |  | 0 |  |  |  |  |  |  |  |  |  |  |  | 2 | 1 | 5131615 | null | null

Вывод таблицы 

1528368781916 | 1 | 1083168000 |  | 10922428129 |  | 2 | 2046573688 | 452049760887694 | 8984049767000887694 | 2 | HPN03 | 400001035 | 000 | 20090511152847 | 20090511152102 |  | 0 | 0 | 1000405100 |  |  |  |  | 20090511152102 | TUDV_POPBGG | 20170801171355 | MIGRATE |  |  |  | MIGRATE |  | 100000 | 1 |  |  |  |  |  |  |  |  |  |  |  | 0 |  |  |  |  |  |  |  |  |  |  |  | 2 | 1 | 5131615 | null | null

Автоматический выбор тем с помощью KSQL

Вы можете использовать KSQL для перераспределения тем.Например, если взять тему вашего источника subscriber, здесь показано, как перераспределить ее с помощью KSQL, чтобы установить ключ: 

ksql> CREATE STREAM SUBSCRIBER_KEYED AS SELECT * FROM str_subscriber_json PARTITION BY sub_id;

 Message
----------------------------
 Stream created and running
----------------------------
ksql>

Это заполняет тему Кафки (SUBSCRIBER_KEYED), по которой вы можете затемопределить таблицу: 

CREATE TABLE subscriber_table (sub_id VARCHAR,contract_id VARCHAR,cust_id VARCHAR,account_id VARCHAR,telecom_service_id VARCHAR,isdn VARCHAR,imsi VARCHAR,serial VARCHAR,status VARCHAR,product_code VARCHAR,offer_id VARCHAR,act_status VARCHAR,sta_datetime BIGINT,active_datetime BIGINT,sub_type VARCHAR,end_datetime BIGINT,expire_datetime BIGINT,shop_code VARCHAR,dev_staff_id VARCHAR,promotion_code VARCHAR,vip VARCHAR,account VARCHAR,create_datetime BIGINT,create_user VARCHAR,update_datetime BIGINT,update_user VARCHAR,deposit VARCHAR,limit_usage VARCHAR,password VARCHAR,org_product_code VARCHAR,num_reset_zone VARCHAR,start_money VARCHAR,is_info_completed VARCHAR,channel_type_id VARCHAR,first_connect VARCHAR,speed VARCHAR,keep_alive VARCHAR,price_plan VARCHAR,local_price_plan VARCHAR,project VARCHAR,local_speed VARCHAR,technology VARCHAR,network_class VARCHAR,infrastructure_type VARCHAR,deploy_accept_date BIGINT,group_id VARCHAR,ip_static VARCHAR,ip_view VARCHAR,ip_lan VARCHAR,ip_wan VARCHAR,ip_gateway VARCHAR,ip_router VARCHAR,num_of_computer VARCHAR,vlan VARCHAR,up_link VARCHAR,private_ip VARCHAR,pay_type VARCHAR,update_number_1 VARCHAR,reg_type_id VARCHAR,type VARCHAR,update_number VARCHAR) \
WITH (KAFKA_TOPIC='SUBSCRIBER_KEYED', VALUE_FORMAT='JSON', KEY = 'sub_id' );

Теперь, когда вы отправляете сообщение на subscriber, даже если оно не введено, таблица будет работать: 

ksql> select * from subscriber_table;
1528369407576 | 1083168000 | 1083168000 |  | 10922428129 |  | 2 | 2046573688 | 452049760887694 | 8984049767000887694 | 2 | HPN03 | 400001035 | 000 | 20090511152847 | 20090511152102 |  | 0 | 0 | 1000405100 |  |  |  |  | 20090511152102 | TUDV_POPBGG | 20170801171355 | MIGRATE |  |  |  | MIGRATE |  | 100000 | 1 |  |  |  |  |  |  |  |  |  |  |  | 0 |  |  |  |  |  |  |  |  |  |  |  | 2 | 1 | 5131615 | null | null
...