Python clickhouse-driver 类库使用学习总结

# -*- coding:utf-8 -*-

import clickhouse_driver

if __name__ == '__main__':
    host = '192.168.88.131'
    port = 9000 # 注意，不能使用默认的8123
    username = 'testacc'
    password = 'test1234'
    database = 'default'

    # 连接方式1
    # conn = clickhouse_driver.connect(database = database,
    #                                  user = username,
    #                                  password=password,
    #                                  host = host,
    #                                  port = port)

    # 连接方式2
    connection_str = f'clickhouse://{username}:{password}@{host}:{port}/{database}'
    conn = clickhouse_driver.connect(connection_str)
    cursor = conn.cursor()

    cursor.execute('SHOW TABLES')
    res = cursor.fetchall()
    print(res) # 输出形如 [('table1',), ('test',)]

    # 删除表
    cursor.execute('DROP TABLE IF EXISTS test')
    print(cursor.fetchall()) # 输出:[]

    cursor.execute('CREATE TABLE test (x Int32) ENGINE = Memory')
    print(cursor.fetchall()) # 输出:[]
    #
    cursor.executemany('INSERT INTO test (x) VALUES', [{'x': 100}])
    print(cursor.rowcount) # 获取execute* 产生记录数 输出：1
    #
    cursor.executemany('INSERT INTO test (x) VALUES', [[200]])
    print(cursor.rowcount)  # 输出：1
    #
    cursor.execute('INSERT INTO test (x) SELECT * FROM system.numbers LIMIT %(limit)s', {'limit': 3})
    print(cursor.rowcount)  # 输出：1

    cursor.execute('SELECT sum(x) AS sum_value FROM test')
    print(cursor.rowcount)  # 输出：1
    print(cursor.columns_with_types) # 获取查询列名及类型，输出：[('sum_value', 'Int64')]

    cursor.execute('SELECT * FROM test')
    print(cursor.rowcount)  # 输出：5
    print(cursor.columns_with_types) # 输出：[('x', 'Int32')]

    res = cursor.fetchall()
    print(res) # 输出：[(100,), (200,), (0,), (1,), (2,)]

    print(cursor.fetchone()) #输出：None

    #############################
    cursor.execute('SELECT * FROM test')
    print(cursor.fetchone())  # 输出：(100,)

    # 仅取3条
    print(cursor.fetchmany(3))  # 输出：[(200,), (0,), (1,)]

    #############################
    cursor.execute('SELECT * FROM test')
    print(cursor.fetchone())  # 输出：(100,)

    print(cursor.fetchall())  # 输出：[(200,), (0,), (1,), (2,)]

    cursor.close()
    conn.close()

conn = clickhouse_driver.connect(connection_str)

'clickhouse://[{username}:{password}@]{host}[:{port}][/{database}]'