招聘通统计分析 2.0：从实时查询到 OLAP 预聚合的性能跃迁

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

CREATE TABLE stats.events (
    event_id String,
    event_type String,       -- view, apply, interview, download
    company_id Int32,
    job_id Int32,
    user_id Int32,
    channel String,          -- pc, app, miniprogram
    city_id Int32,
    source String,           -- search, recommend, direct
    created_at DateTime64(3)
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(created_at)
ORDER BY (company_id, event_type, created_at)
TTL created_at + INTERVAL 2 YEAR;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

CREATE TABLE stats.daily_company (
    company_id Int32,
    event_date Date,
    event_type String,
    pv AggregateFunction(count, String),
    uv AggregateFunction(uniq, Int32)
) ENGINE = AggregatingMergeTree()
PARTITION BY toYYYYMM(event_date)
ORDER BY (company_id, event_date, event_type);

-- Materialized View 自动同步
CREATE MATERIALIZED VIEW stats.daily_company_mv TO stats.daily_company AS
SELECT
    company_id,
    toDate(created_at) AS event_date,
    event_type,
    countState(event_id) AS pv,
    uniqState(user_id) AS uv
FROM stats.events
GROUP BY company_id, event_date, event_type;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

CREATE TABLE stats.daily_job (
    company_id Int32,
    job_id Int32,
    event_date Date,
    event_type String,
    pv AggregateFunction(count, String),
    uv AggregateFunction(uniq, Int32)
) ENGINE = AggregatingMergeTree()
PARTITION BY toYYYYMM(event_date)
ORDER BY (company_id, job_id, event_date, event_type);

CREATE MATERIALIZED VIEW stats.daily_job_mv TO stats.daily_job AS
SELECT
    company_id, job_id,
    toDate(created_at) AS event_date,
    event_type,
    countState(event_id) AS pv,
    uniqState(user_id) AS uv
FROM stats.events
GROUP BY company_id, job_id, event_date, event_type;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

class StatsQueryService
{
    public function query($params)
    {
        // 根据查询维度选择预聚合表
        $dimensions = $this->getDimensions($params);
        
        if ($this->canUseDailyCompany($dimensions)) {
            return $this->queryFromDailyCompany($params);
        }
        if ($this->canUseDailyJob($dimensions)) {
            return $this->queryFromDailyJob($params);
        }
        // 没有合适的预聚合表，查明细表
        return $this->queryFromEvents($params);
    }

    private function queryFromDailyCompany($params)
    {
        $sql = "SELECT
                    event_date,
                    countMerge(pv) as pv,
                    uniqMerge(uv) as uv
                FROM stats.daily_company
                WHERE company_id = ? 
                  AND event_date BETWEEN ? AND ?
                  AND event_type = ?
                GROUP BY event_date
                ORDER BY event_date";
        
        return $this->clickhouse->query($sql, [
            $params['company_id'],
            $params['start_date'],
            $params['end_date'],
            $params['event_type'],
        ]);
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

CREATE TABLE stats.events_kafka (
    event_id String,
    event_type String,
    company_id Int32,
    job_id Int32,
    user_id Int32,
    channel String,
    city_id Int32,
    source String,
    created_at DateTime64(3)
) ENGINE = Kafka()
SETTINGS
    kafka_broker_list = 'kafka1:9092,kafka2:9092,kafka3:9092',
    kafka_topic_list = 'stats_events',
    kafka_group_name = 'clickhouse_stats',
    kafka_format = 'JSONEachRow',
    kafka_num_consumers = 3;

-- Materialized View 把 Kafka 数据写入明细表
CREATE MATERIALIZED VIEW stats.events_consumer TO stats.events AS
SELECT * FROM stats.events_kafka;