ClickHouseの使い方とパーティションの話

この記事は、[ただの集団のアドベントカレンダー](https://adventar.org/calendars/3447)の24日目の記事になります。2018年も残すところ僅かですが、アドベントカレンダーもいよいよ大詰めとなりました。
私自身、アドベントカレンダーの記事として書くのは初めてになります。よろしくお願いします。

# はじめに
Webエンジニアとして開発業務を行っていますが、データ分析基盤についても触れる機会が多いです。
データ分析基盤といえばマネージドサービスであるAmazon RedshiftやBigQueryが有名ですが、今回はClickHouseについて紹介したいと思います。

## ClickHouseとは
列指向型のDBです。公式ドキュメント: https://clickhouse.yandex/
ソースコードは[GitHub](https://github.com/yandex/ClickHouse)で公開されており、C++で実装されています。

## きっかけ
1年ほど前の勉強会で、[GENIEEさんの記事](https://tech.geniee.co.jp/entry/2017/07/20/160100)を見てからClickHouseについて興味を抱きました。とはいえ、いろいろとわからないところもあるため勉強しながら理解を深めています。

## ClickHouseの特徴
[公式のベンチマーク](https://clickhouse.yandex/benchmark.html)でも謳われているように、__パフォーマンスが売り__です。多彩なテーブルエンジンを提供しており、用途に応じて使い分けると十二分にパフォーマンスを発揮してくれると思います。

最大のデメリットは、__行のUPDATEやDELETEができない(= INSERTのみ)__という点です。故にログのデータベースとしての用途が適しているのですが、パーティションの設計を考えないと運用で苦しむことになります。
また、__UNIQUEキー制約に該当する機能もありません__。重複を防ぐには、別で実装してやる必要があります。

## データ型
メジャーどころのデータ型は全て揃っています。詳しくは[公式ドキュメント](https://clickhouse.yandex/docs/en/data_types/)を参照。
注意点としては、__DateTime型はミリ秒以下の値を持てない__、という点でしょうか。

## SQL
基本構文は、標準SQLとほぼ一緒です。私は普段はMySQLをメインで使用していますが、特に違和感なくSQLを書けています。SELECT句やGROUP BY句で列別名も使えます。
少し扱いにくいと思うところは、[__SELECT句でのJOINが2つのテーブルにしかできない__](https://clickhouse.yandex/docs/en/query_language/select/#join-clause)点でしょうか。

## テーブルエンジン
詳しくは[公式ドキュメント](https://clickhouse.yandex/docs/en/operations/table_engines)を参照。今回は以下で[MergeTree Engine](https://clickhouse.yandex/docs/en/operations/table_engines/mergetree/)を取り上げます。

## MergeTree Engineとパーティション
※こちらの記事も参考にさせていただきました。 https://qiita.com/mikage/items/23da072340843a6c24e9 

ClickHouseは行のUPDATE・DELETEができないため、データを修正したい場合は工夫が必要です。
時系列でデータを保持する運用の場合、以前は以下のような構成でしかできないと考えていました。
日付ごとにMergeTree Engineのテーブルを作り、[Merge Engine](https://clickhouse.yandex/docs/en/operations/table_engines/merge/)を使って単一のテーブルで参照するイメージです。

```sql
create table score_20181101 (
    dt date,
    name String,
    group String,
    score UInt16
) ENGINE=MergeTree(dt, (dt, name), 8192);

create table score (
    dt date,
    name String,
    group String,
    score UInt16
) ENGINE=Merge(test, '^score_');
```

データを修正したい場合は、テーブルをDROPしてからもう一回作り直す要領です。
しかし、現在のバージョンでは任意のパーティションを指定することができるようで、もっと効率よくできそう...ただ、実際はどうなんだろう？と思ってテストデータを用意して調べることにしました。

## 環境構築
[公式のdockerイメージ](https://hub.docker.com/r/yandex/clickhouse-server/)が提供されていますので、それを使います。
今回はdocker-composeを使って進めてみます。

docker-compose.ymlの中身

```yml
version: "3.7"

services:
  client:
    image: yandex/clickhouse-client
    entrypoint:
      - /bin/sleep
    command:
      - infinity
  server:
    image: yandex/clickhouse-server
    ports:
      - 8123:8123
    volumes:
      - ./volume:/var/lib/clickhouse
```

起動し、clickhouse-clientで接続します。

```sh
$ docker-compose up -d
$ docker exec -it clickhouse_client_1 /usr/bin/clickhouse-client --host clickhouse_server_1 --multiline
ClickHouse client version 1.1.54390.
Connecting to clickhouse_server_1:9000 as user default.
Connected to ClickHouse server version 1.1.54390.

b5cf859468ab :)
```

今回はclickhouse-clientを使いますが、clickhouse-serverはHTTP等の[インターフェース](https://clickhouse.yandex/docs/en/interfaces/)も提供されているので、必要に応じてそちらを使うこともできます。

### データベースを作る
初期状態では、defaultデータベースとなっています。

```sh
b5cf859468ab :) show databases;

┌─name────┐
│ default │
│ system  │
└─────────┘
```

今回はtestデータベースを使います。

```sh
b5cf859468ab :) create database test;

CREATE DATABASE test

Ok.

0 rows in set. Elapsed: 0.017 sec.

b5cf859468ab :) use test;

USE test

Ok.

0 rows in set. Elapsed: 0.002 sec.
```

## パーティションのパターン
パターンを3つに分け、日ごと・週ごと・月ごとでパーティションを設定してみます。

- xcore: 日ごと
- ycore: 週ごと(toMonday関数を用いた月曜日を起点とする)
- zcore: 月ごと

```sql
create table xcore (
    dt date,
    name String,
    group String,
    score UInt16
) ENGINE=MergeTree() PARTITION BY dt ORDER BY (dt,name) SETTINGS index_granularity=8192;

create table ycore (
    dt date,
    name String,
    group String,
    score UInt16
) ENGINE=MergeTree() PARTITION BY toMonday(dt) ORDER BY (dt,name) SETTINGS index_granularity=8192;

create table zcore (
    dt date,
    name String,
    group String,
    score UInt16
) ENGINE=MergeTree() PARTITION BY toYYYYMM(dt) ORDER BY (dt,name) SETTINGS index_granularity=8192;
```

テストデータとなるCSVを作ってみます。テストデータは私の思いつきです。

```python
#!/usr/bin/env python

import sys, csv, random
from datetime import date, timedelta


def date_range():
    base = date(2018, 11, 1)
    return [base + timedelta(days=x) for x in range(0, 30)]


def write_csv(dt, rows):
    file_path = f'./csv/{date.strftime(dt, "%Y%m%d")}.csv'
    with open(file_path, 'w') as f:
        writer = csv.writer(f)
        writer.writerows(rows)


persons = [
    ('person1', 'A'),
    ('person2', 'A'),
    ('person3', 'A'),
    ('person4', 'B'),
    ('person5', 'B'),
    ('person6', 'B'),
    ('person7', 'C'),
    ('person8', 'C'),
    ('person9', 'C')
]


def make_rows(dt_str):
    p = random.choice(persons)
    rows = [dt_str, p[0], p[1], random.randint(1, 100)]
    return rows


if __name__ == '__main__':
    for dt in date_range()):
        dt_str = date.strftime(dt, "%Y-%m-%d")
        rows = [make_rows(dt_str) for i in range(0, 1000000)]
        write_csv(dt, rows)
```

データをINSERTします。ycore・zcoreテーブルも同様の手順です。

```sh
$ cat csv/201811*.csv | docker exec -i clickhouse_client_1 /usr/bin/clickhouse-client --host clickhouse_server_1 --query="INSERT INTO test.xcore FORMAT CSV"
```

パフォーマンス確認クエリ)

```sql
SELECT
    dt,
    name,
    count(1) AS cnt,
    sum(score) AS score_sum,
    avg(score) AS score_avg,
    median(score) AS score_med
FROM xcore
GROUP BY
    dt,
    name
ORDER BY
    dt ASC,
    name ASC

(結果は省略)
270 rows in set. Elapsed: 3.238 sec. Processed 30.00 million rows, 600.00 MB (9.27 million rows/s., 185.33 MB/s.)
```

パーティション確認

```sql
SELECT
    table,
    partition,
    sum(rows) AS rows,
    sum(bytes) AS bytes
FROM system.parts
WHERE table LIKE '_core'
GROUP BY
    table,
    partition
ORDER BY
    table ASC,
    partition ASC

┌─table─┬─partition────┬─────rows─┬────bytes─┐
│ xcore │ '2018-11-01' │  1000000 │  1592821 │
│ xcore │ '2018-11-02' │  1000000 │  1593427 │
│ xcore │ '2018-11-03' │  1000000 │  1593207 │
│ xcore │ '2018-11-04' │  1000000 │  1593657 │
│ xcore │ '2018-11-05' │  1000000 │  1593181 │
│ xcore │ '2018-11-06' │  1000000 │  1593871 │
│ xcore │ '2018-11-07' │  1000000 │  1593121 │
│ xcore │ '2018-11-08' │  1000000 │  1596000 │
│ xcore │ '2018-11-09' │  1000000 │  1593858 │
│ xcore │ '2018-11-10' │  1000000 │  1594210 │
│ xcore │ '2018-11-11' │  1000000 │  1593228 │
│ xcore │ '2018-11-12' │  1000000 │  1594379 │
│ xcore │ '2018-11-13' │  1000000 │  1593403 │
│ xcore │ '2018-11-14' │  1000000 │  1595158 │
│ xcore │ '2018-11-15' │  1000000 │  1593774 │
│ xcore │ '2018-11-16' │  1000000 │  1594756 │
│ xcore │ '2018-11-17' │  1000000 │  1593107 │
│ xcore │ '2018-11-18' │  1000000 │  1595120 │
│ xcore │ '2018-11-19' │  1000000 │  1593995 │
│ xcore │ '2018-11-20' │  1000000 │  1594450 │
│ xcore │ '2018-11-21' │  1000000 │  1592488 │
│ xcore │ '2018-11-22' │  1000000 │  1592338 │
│ xcore │ '2018-11-23' │  1000000 │  1592955 │
│ xcore │ '2018-11-24' │  1000000 │  1594515 │
│ xcore │ '2018-11-25' │  1000000 │  1594216 │
│ xcore │ '2018-11-26' │  1000000 │  1594527 │
│ xcore │ '2018-11-27' │  1000000 │  1593790 │
│ xcore │ '2018-11-28' │  1000000 │  1594172 │
│ xcore │ '2018-11-29' │  1000000 │  1594139 │
│ xcore │ '2018-11-30' │  1000000 │  1592715 │
│ ycore │ '2018-10-29' │  4000000 │  6366752 │
│ ycore │ '2018-11-05' │  7000000 │ 11138319 │
│ ycore │ '2018-11-12' │  7000000 │ 11143072 │
│ ycore │ '2018-11-19' │  7000000 │ 11140452 │
│ ycore │ '2018-11-26' │  5000000 │  7956492 │
│ zcore │ 201811       │ 30000000 │ 47736183 │
└───────┴──────────────┴──────────┴──────────┘

SELECT
    table,
    sum(rows) AS rows,
    sum(bytes) AS bytes
FROM system.parts
WHERE table LIKE '_core'
GROUP BY table
ORDER BY table ASC

┌─table─┬─────rows─┬────bytes─┐
│ xcore │ 30000000 │ 47814578 │
│ ycore │ 30000000 │ 47745087 │
│ zcore │ 30000000 │ 47736183 │
└───────┴──────────┴──────────┘
```

パーティションの削除

```sql
ALTER TABLE test.xcore DROP PARTITION '2018-11-01';
```

データも削除されたことが確認できます。

```sql
SELECT min(dt)
FROM test.xcore

┌────min(dt)─┐
│ 2018-11-02 │
└────────────┘
```

## Mergeエンジンとの併用
パーティションが異なっても、Mergeエンジンを使ってまとめた結果を出すことはできるようです　(当然、クエリ実行の時間はかかりますが)。

```sql
CREATE TABLE test.core (
    dt Date,
    name String,
    group String,
    score UInt16
) ENGINE = Merge(test, 'core$')

SELECT
    dt,
    name,
    count(1) AS cnt,
    sum(score) AS score_sum,
    avg(score) AS score_avg,
    median(score) AS score_med
FROM core
GROUP BY
    dt,
    name
ORDER BY
    dt ASC,
    name ASC

(結果は省略)
270 rows in set. Elapsed: 9.975 sec. Processed 89.00 million rows, 1.78 GB (8.92 million rows/s., 178.45 MB/s.)
```

## 検証結果
- ひと月分(100万レコード/日)でも参照時のパフォーマンスでは少し差が出ている。塵も積もれば山となりそう。
- テストデータで確認した限りでは、データサイズは殆ど変わらない。データ数・データサイズが増えると変わるのだろうか。
- 運用時は、パーティションを段階に分けて使うとデータが増えてもパフォーマンスが上がりそう。
  - 直近のデータはパーティションを細かく、昔のデータはパーティションを大きく設定する、とか。オンラインでのテーブル切り替えは[RENAME](https://clickhouse.yandex/docs/en/query_language/misc/#rename)を使えばできそう

## まとめ
ClickHouseは少し癖がありますがトリッキーなことができたりと、使ってみると面白いデータベースです。
以前は日本語の文献も少なかったですがこの1年で徐々に増えてきているようで、 (拙い英語力である) 私には非常に助けられています。個人的にも、もっと広まって欲しいと思っています。