本文主要是介绍Python3 使用 clickhouse_driver 操作 clickhouse，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！
'''
版本：
Python 3.7 x86
clickhouse 24.6.1.3573
clickhouse-driver 0.2.7
'''
代码一：
from clickhouse_driver import Client# 准备参数
host = "192.168.1.112"
port = 9000
username = "default"
password = "123456"
database = "default"# client = Client('localhost')
client = Client(host=host,port=port,database=database,user=username,password=password)result = client.execute('SHOW TABLES')
print(type(result))
for table in result:print(table)
client.execute('DROP TABLE IF EXISTS test')
client.execute('CREATE TABLE test (x Int32) ENGINE = Memory')
client.execute('INSERT INTO test (x) VALUES',[{'x': 100}])
client.execute('INSERT INTO test (x) VALUES', [[200]])
client.execute('INSERT INTO test (x) ''SELECT * FROM system.numbers LIMIT %(limit)s',{'limit': 3})
sumdata = client.execute('SELECT sum(x) FROM test')
print(type(sumdata))
for i in sumdata:for k in i:print(k)'''
输出：
<class 'list'>
('my_table',)
('new_table',)
('test',)
('user',)
<class 'list'>
303
'''代码二：
from clickhouse_driver import connect# 准备参数
host = "192.168.1.112"
port = 9000
username = "default"
password = "123456"
database = "default"# conn = connect('clickhouse://localhost')
conn = connect(host=host,port=port,database=database,user=username,password=password)cursor = conn.cursor()cursor.execute('SHOW TABLES')
print(cursor.fetchall())cursor.execute('DROP TABLE IF EXISTS test')
cursor.fetchall()cursor.execute('CREATE TABLE test (x Int32) ENGINE = Memory')
cursor.fetchall()cursor.executemany('INSERT INTO test (x) VALUES',[{'x': 100}])
cursor.rowcountcursor.executemany('INSERT INTO test (x) VALUES', [[200],[300]])
cursor.rowcountcursor.execute('INSERT INTO test (x)''SELECT * FROM system.numbers LIMIT %(limit)s',{'limit': 3})
cursor.rowcountcursor.execute('SELECT * FROM test')
result = cursor.fetchall()
print(type(result))
for i in result:print(i)'''
输出：
[('my_table',), ('new_table',), ('test',), ('user',)]
<class 'list'>
(100,)
(200,)
(300,)
(0,)
(1,)
(2,)
'''
代码三：
from clickhouse_driver import connect# 准备参数
host = "192.168.1.112"
port = 9000
username = "default"
password = "123456"
database = "default"# conn = connect('clickhouse://localhost')
conn = connect(host=host,port=port,database=database,user=username,password=password)# 创建游标
cursor = conn.cursor()cursor.execute('SHOW TABLES')
print(cursor.fetchall())# 删除表
cursor.execute('drop table if exists my_table')create_table_sql = "CREATE TABLE IF NOT EXISTS my_table (id Int32, name String, age Int32) ENGINE = MergeTree() ORDER BY id"# 执行创建数据表的 SQL 语句
cursor.execute(create_table_sql)# 批量插入数据
# insert_sql = "INSERT INTO my_table (id,name,age) VALUES (%(id)s,%(name)s,%(age)s)"
insert_sql = "INSERT INTO my_table (id,name,age) VALUES "'''
data = [{'id':1, 'name':'Alice', 'age':20},{'id':2, 'name':'Alice', 'age':20},{'id':3, 'name':'Alice', 'age':20},
]
'''
#
data = [(1,'Alice',20),(2,'Alice',20),(3,'Alice',20),
]# 执行批量插入数据的 SQL 语句
cursor.executemany(insert_sql, data)# 提交事务
conn.commit()cursor.execute('SELECT id FROM my_table')
result = cursor.fetchall()
print(type(result))
for i in result:print(i)# 关闭连接
conn.close()
'''
参考：
https://blog.51cto.com/u_16175492/8291625
https://blog.51cto.com/u_16213366/9389145
https://blog.51cto.com/u_16213301/8845983
https://clickhouse-driver.readthedocs.io/en/latest/quickstart.html
https://pypi.org/project/clickhouse-driver/
'''
这篇关于Python3 使用 clickhouse_driver 操作 clickhouse的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
