Python使用ch-orm對ClickHouse簡單查詢及寫入

引言

前不久新項目中需要用到ClickHouse，作為一個合格的Python程式員，首先當然是找找有沒有合適的輪子。

翻了一圈，infi.clickhouse_orm在功能和易用性上沒有明顯的短板，其ORM API對後端程式員格外親切。可惜主分支已經八個月沒有更新了，據聞核心開發者已離職，而infi.clickhouse_orm尚不支持一些我需要的新功能如Geo類型和函數，基於這些原因，這篇文章的主角ch-orm也就誕生了。

ch-orm庫fork自infi.clickhouse_orm(v2.1.1)。

與infi相比，ch-orm支持同步和非同步兩種方式與ClickHouse伺服器交互，它添加了一些新功能：

• 非同步支持(AioDatabase)

  • 為所有同步API提供async介面

• 類型註解

  • 大部分對外API實現了類型註解

• 新的類型支持

  • Tuple
  • Geo類型；Point、Ring等

• 新的函數支持

  • Geo函數等

• 支持創建臨時表(TemporaryModel)

  • session會話

需要提醒的是，ch-orm僅使用ClickHouse的http協議，不支持TCP協議。更多細節參見Github上的文檔。

快速開始

1. 安裝

通過pip安裝ch-orm

pip install ch-orm

2. 定義一個模型

雖然pypi的庫名為ch-orm，但在代碼中需要導入的是clickhouse_orm。

from clickhouse_orm import Database, Model, MergeTree
from clickhouse_orm.fields import (
    StringField, Int32Field, UUIDField, Int8Field
)
from clickhouse_orm.contrib.geo.fields import PointField

class Residence(Model):
    uuid = UUIDField()
    residence_type = Int8Field()
    geo = PointField(db_column='geo_wgs84')
    geohash_wgs84 = StringField()
    province = StringField()
    city = StringField()
    district = StringField()
    poi_id = Int32Field(default=1000)
    poi_name = StringField()
    p_geo_bd09 = PointField()

    engine = MergeTree(partition_key=('uuid', ), order_by=('uuid', ))

    @classmethod
    def table_name(cls):
        return 'residence'

我們定義了一個Residence模型，它將會映射到ClickHouse上的residence表，而Residence中眾多Field屬性則被映射為表中的列，可以在Python中對Residence實例進行操作進而處理ClickHouse（沒錯，就像Django ORM所做的那樣）

接下來，先假定此時residence尚不存在，藉助Residence來創建它。

創建數據表

想要對資料庫執行操作，首先必須實例化一個Database對象(或AioDatabase)，可以粗淺的理解為它和資料庫連接屬於一類抽象，內部實現對後端資料庫的交互。

from clickhouse_orm.database import Database
from clickhouse_orm.aio.database import AioDatabase

# 以同步方式創建資料庫
sync_db = Database('db-test', db_url='http://localhost:8123/')
sync_db.create_table(Residence)


# 以非同步方式創建資料庫
async def main():
    async_db = AioDatabase('db-test', db_url='http://localhost:8123/')
    # 非同步模型下需要主動執行init方法初始化
    await async_db.init()
    await async_db.create_table(Residence)

此時，db-test庫內應當出現了一個名為residence的表。

插入數據

ClickHouse在數據寫入性能表現十分優異，ch-orm能輕易處理寫入數據需求

以寫入100萬條數據為例，使用生成器創建100萬個Residence隨機實例

import uuid

from clickhouse_orm.contrib.geo.fields import Point

# 同步寫入100萬條residence
sync_db.insert(
    (Residence(uuid=str(uuid.uuid4()), geo=Point(120, 20)) for _ in range(1000000)),
    batch_size=10000
)


# 非同步寫入100萬條residence
async def insert():
    ...
    await async_db.insert(
        (Residence(uuid=str(uuid.uuid4()), geo=Point(120, 20)) for _ in range(1000000)),
        batch_size=10000
    )

示例中我們僅對uuid和geo列進行賦值，其他欄位會被設置為預設值（而非None值）

可以看看residence表中有多少條數據

# 同步方式查詢Residence行數
Residence.objects_in(sync_db).count()


# 非同步方式查詢Residence行數
async def read_count():
    ...
    await Residence.objects_in(async_db).count()

查詢API

ch-orm實現了QuerySet，暴露API基本參照Django設計的，如前述的獲取表行數的count()方法就來自QuerySet。

與Django不同的是，ch-orm僅將QuerySet作為查詢實例，不具備查詢結果緩存功能，這代表如果對一個QuerySet對象執行兩次迭代，與後端資料庫的交互將變成兩次而非一次。

可以通過Model的類方法objects_in獲得一個QuerySet實例，接著來查詢uuid="48d75e4d-8e6f-4acd-a2e9-f4c3059b5b30"的數據

# 同步API
queryset = Residence.objects_in(sync_db)
queryset = queryset.filter(Residence.uuid == "48d75e4d-8e6f-4acd-a2e9-f4c3059b5b30")
result = list(queryset)

# 對於非同步API
queryset = Residence.objects_in(async_db)
queryset = queryset.filter(Residence.uuid == "48d75e4d-8e6f-4acd-a2e9-f4c3059b5b30")
result = [_ async for _ in queryset]

真正的查詢請求是在對queryset迭代時處理的，因此下列兩行代碼不會與資料庫後端進行交互

queryset = Residence.objects_in(sync_db)
queryset = queryset.filter(Residence.uuid == "48d75e4d-8e6f-4acd-a2e9-f4c3059b5b30")

最終得到一個由Residence實例的組成的結果列表result。

3. 略微複雜功能

ch-orm具備日常使用的大多數場景功能

• 執行原生查詢並創建動態對象

• 現有表生成模型類

• F函數

• Q查詢

• Aggregation 聚合查詢

• order_by 查詢排序

• distinct 結果去重

• Pagination 查詢分頁

• 表引擎

• ...

這些內容Github倉庫有相應的文檔，限於本文篇幅這裡就不再過多介紹。