招聘通统计分析后台架构：从单表查询到独立统计服务的演进之路

 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

public function actionJobStats()
{
    $companyId = Yii::$app->user->company_id;
    $startDate = Yii::$app->request->get('start_date');
    $endDate = Yii::$app->request->get('end_date');

    // 职位浏览量
    $views = (new Query())
        ->from('job_view_log')
        ->where(['company_id' => $companyId])
        ->andWhere(['between', 'created_at', strtotime($startDate), strtotime($endDate)])
        ->count();

    // 简历投递数
    $applies = (new Query())
        ->from('job_apply')
        ->where(['company_id' => $companyId])
        ->andWhere(['between', 'created_at', strtotime($startDate), strtotime($endDate)])
        ->count();

    // 面试数
    $interviews = (new Query())
        ->from('interview')
        ->where(['company_id' => $companyId])
        ->andWhere(['between', 'created_at', strtotime($startDate), strtotime($endDate)])
        ->count();

    return [
        'views' => $views,
        'applies' => $applies,
        'interviews' => $interviews,
        'apply_rate' => $views > 0 ? round($applies / $views * 100, 2) : 0,
        'interview_rate' => $applies > 0 ? round($interviews / $applies * 100, 2) : 0,
    ];
}

1
2
3
4
5
6
7

// 统计查询用从库
$db = Yii::$app->dbSlave;
$views = (new Query())
    ->from('job_view_log')
    ->where(['company_id' => $companyId])
    ->andWhere(['between', 'created_at', $start, $end])
    ->count($db);

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

CREATE TABLE `stats_daily_company` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `company_id` int(11) NOT NULL,
  `date` date NOT NULL,
  `views` int(11) NOT NULL DEFAULT '0',
  `applies` int(11) NOT NULL DEFAULT '0',
  `interviews` int(11) NOT NULL DEFAULT '0',
  `offers` int(11) NOT NULL DEFAULT '0',
  PRIMARY KEY (`id`),
  UNIQUE KEY `company_date` (`company_id`,`date`),
  KEY `date` (`date`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

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

public function actionAggregateDaily()
{
    $yesterday = date('Y-m-d', strtotime('-1 day'));
    $start = strtotime($yesterday . ' 00:00:00');
    $end = strtotime($yesterday . ' 23:59:59');

    // 按公司统计浏览量
    $viewStats = (new Query())
        ->select(['company_id', 'COUNT(*) as cnt'])
        ->from('job_view_log')
        ->where(['between', 'created_at', $start, $end])
        ->groupBy('company_id')
        ->all();

    foreach ($viewStats as $stat) {
        $this->upsertDailyStats($stat['company_id'], $yesterday, ['views' => $stat['cnt']]);
    }

    // 同理统计投递、面试等
    // ...
}

1
2
3
4
5
6

$stats = (new Query())
    ->select(['SUM(views) as views', 'SUM(applies) as applies', 'SUM(interviews) as interviews'])
    ->from('stats_daily_company')
    ->where(['company_id' => $companyId])
    ->andWhere(['between', 'date', $startDate, $endDate])
    ->one();

1
2
3
4
5

业务系统 → 发消息到 Kafka → 统计服务消费 → 写入 ClickHouse
                              ↓
                        实时聚合（Redis）
                              ↓
统计查询 API ← 读 ClickHouse / Redis

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

CREATE TABLE stats.events (
    event_id String,
    event_type String,       -- view, apply, interview, offer
    company_id Int32,
    job_id Int32,
    user_id Int32,
    channel String,          -- pc, app, miniprogram
    city_id Int32,
    created_at DateTime
) 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

// 消费 Kafka 消息，更新 Redis 实时计数
$key = "stats:realtime:{$companyId}:{$eventType}:{$date}";
$redis->incr($key);

// 同时异步写入 ClickHouse
$clickhouse->insert('stats.events', $eventData);

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

public function queryStats($companyId, $startDate, $endDate, $dimensions = [])
{
    $today = date('Y-m-d');
    $result = [];

    // 历史数据查 ClickHouse
    if ($startDate < $today) {
        $histEnd = min($endDate, date('Y-m-d', strtotime('-1 day')));
        $result = $this->clickhouseQuery($companyId, $startDate, $histEnd, $dimensions);
    }

    // 当天数据查 Redis
    if ($endDate >= $today) {
        $realtime = $this->redisQuery($companyId, $today, $dimensions);
        $result = $this->mergeStats($result, $realtime);
    }

    return $result;
}