前不久新项目中需要用到ClickHouse,作为一个合格的Python程序员,首先当然是找找有没有合适的轮子。
都匀ssl适用于网站、小程序/APP、API接口等需要进行数据传输应用场景,ssl证书未来市场广阔!成为创新互联公司的ssl证书销售渠道,可以享受市场价格4-6折优惠!如果有意向欢迎电话联系或者加微信:13518219792(备注:SSL证书合作)期待与您的合作!
翻了一圈,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)
类型注解
新的类型支持
新的函数支持
支持创建临时表(TemporaryModel)
需要提醒的是,ch-orm仅使用ClickHouse的http协议,不支持TCP协议。更多细节参见Github上的文档。
通过pip安装ch-orm
pip install ch-orm
虽然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()),
batch_size=
)
# 异步写入100万条residence
async def insert():
...
await async_db.insert(
(Residence(uuid=str(uuid.uuid4()), geo=Point(120, 20)) for _ in range()),
batch_size=
)
示例中仅对uuid
和geo
列进行赋值,其他字段会被设置为默认值(而非None值)
可以看看residence
表中有多少条数据
# 同步方式查询Residence行数
Residence.objects_in(sync_db).count()
# 异步方式查询Residence行数
async def read_count():
...
await Residence.objects_in(async_db).count()
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。
ch-orm具备日常使用的大多数场景功能
执行原生查询并创建动态对象
现有表生成模型类
F函数
Q查询
Aggregation 聚合查询
order_by 查询排序
distinct 结果去重
Pagination 查询分页
表引擎
...
这些内容Github仓库有相应的文档,限于本文篇幅这里就不再过多介绍。