V7gFSew8

1
2drop table clicker.purchases_consumer on cluster analytics;
3drop table clicker.purchases_queue on cluster analytics;
4drop table clicker.purchases_dist on cluster analytics;
5
6ALTER TABLE clicker.purchases ON CLUSTER analytics ADD COLUMN IF NOT EXISTS event_id String DEFAULT '';
7
8
9CREATE TABLE IF NOT EXISTS clicker.purchases_queue
10ON CLUSTER analytics
11(
12    created_on UInt64,
13    profile_id String,
14    transaction_id String,
15    lot_id String,
16    environment String,
17    lot_string_id String,
18    is_qa_purchase String,
19    state String,
20    additional_id String,
21    currency_code String,
22    price Float64,
23    price_usd Float64,
24    level UInt64,
25    event_id String
26) ENGINE = Kafka()
27SETTINGS
28    kafka_broker_list = 'kafka-int.prod.playful-fairies.com:9092',
29    kafka_topic_list = 'clicker_purchases',
30    kafka_group_name = 'group1',
31    kafka_format = 'JSONEachRow';
32
33
34CREATE TABLE clicker.purchases_dist
35ON CLUSTER analytics AS clicker.purchases
36ENGINE = Distributed(analytics, clicker, purchases, cityHash64(profile_id));
37
38
39CREATE MATERIALIZED VIEW clicker.purchases_consumer
40ON CLUSTER analytics
41TO clicker.purchases_dist
42AS SELECT
43    toDate(created_on) as day,
44    created_on,
45    profile_id,
46    transaction_id,
47    lot_id,
48    environment,
49    lot_string_id,
50    is_qa_purchase,
51    state,
52    additional_id,
53    currency_code,
54    price,
55    price_usd,
56    level,
57    event_id
58FROM clicker.purchases_queue;
59
60
61{"profile_id":"yareeek","playtime":1439764,"transaction_id":"1192262934","lot_id":"cueria10p1328397","environment":"site","lot_string_id":"B100110215","is_qa_purchase":"False","state":"ok","event_id":"1","additional_id":"2327310115","currency_code":"USD","price":4.99,"price_usd":4.99,"created_on":1585155550,"level":25}