ClickHouse Connect 드라이버 API

import clickhouse_connect

client = clickhouse_connect.get_client()
print(client.server_version)
# 출력: '22.10.1.98'

import clickhouse_connect

client = clickhouse_connect.get_client(host='play.clickhouse.com', secure=True, port=443, user='play', password='clickhouse')
print(client.command('SELECT timezone()'))
# 출력: 'Etc/UTC'

import clickhouse_connect

client = clickhouse_connect.get_client(
    host='play.clickhouse.com',
    user='play',
    password='clickhouse',
    port=443,
    session_id='example_session_1',
    connect_timeout=15,
    database='github',
    settings={'distributed_ddl_task_timeout':300},
)
print(client.database)
# 출력: 'github'

import clickhouse_connect

# 시작 시 한 번만 생성
client = clickhouse_connect.get_client(host='my-host', username='default', password='password')

# 모든 쿼리에 재사용
for i in range(1000):
    result = client.query('SELECT count() FROM users')

# 종료 시 닫기
client.close()

# 나쁜 예: 초기화 오버헤드가 큰 클라이언트를 1000개 생성합니다
for i in range(1000):
    client = clickhouse_connect.get_client(host='my-host', username='default', password='password')
    result = client.query('SELECT count() FROM users')
    client.close()

import clickhouse_connect
import threading

# 옵션 1: 세션 비활성화 (공유 클라이언트에 권장)
client = clickhouse_connect.get_client(
    host='my-host',
    username='default',
    password='password',
    autogenerate_session_id=False  # 스레드 안전성을 위해 필요
)

def worker(thread_id):
    # 이제 모든 스레드에서 동일한 클라이언트를 안전하게 사용할 수 있음
    result = client.query(f"SELECT {thread_id}")
    print(f"Thread {thread_id}: {result.result_rows[0][0]}")

threads = [threading.Thread(target=worker, args=(i,)) for i in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()

client.close()
# 출력:
# Thread 0: 0
# Thread 7: 7
# Thread 1: 1
# Thread 9: 9
# Thread 4: 4
# Thread 2: 2
# Thread 8: 8
# Thread 5: 5
# Thread 6: 6
# Thread 3: 3

def worker(thread_id):
    # 각 스레드는 격리된 세션을 가진 독립적인 클라이언트를 사용합니다
    client = clickhouse_connect.get_client(host='my-host', username='default', password='password')
    client.command('CREATE TEMPORARY TABLE temp (id UInt32) ENGINE = Memory')
    # ... 임시 테이블 사용 ...
    client.close()

client = clickhouse_connect.get_client(host='my-host', username='default', password='password')
try:
    result = client.query('SELECT 1')
finally:
    client.close()

with clickhouse_connect.get_client(host='my-host', username='default', password='password') as client:
    result = client.query('SELECT 1')

import datetime

my_date = datetime.datetime(2022, 10, 1, 15, 20, 5)

parameters = {'table': 'my_table', 'v1': my_date, 'v2': "a string with a single quote'"}
client.query('SELECT * FROM {table:Identifier} WHERE date >= {v1:DateTime} AND string ILIKE {v2:String}', parameters=parameters)

SELECT *
FROM my_table
WHERE date >= '2022-10-01 15:20:05'
  AND string ILIKE 'a string with a single quote\''

import datetime

my_date = datetime.datetime(2022, 10, 1, 15, 20, 5)

parameters = {'v1': my_date, 'v2': "a string with a single quote'"}
client.query('SELECT * FROM my_table WHERE date >= %(v1)s AND string ILIKE %(v2)s', parameters=parameters)

SELECT *
FROM my_table
WHERE date >= '2022-10-01 15:20:05'
  AND string ILIKE 'a string with a single quote\''

import ipaddress

parameters = (35200.44, ipaddress.IPv4Address(0x443d04fe))
client.query('SELECT * FROM some_table WHERE metric >= %s AND ip_address = %s', parameters=parameters)

SELECT *
FROM some_table
WHERE metric >= 35200.44
  AND ip_address = '68.61.4.254''

settings = {'merge_tree_min_rows_for_concurrent_read': 65535,
            'session_id': 'session_1234',
            'use_skip_indexes': False}
client.query("SELECT event_type, sum(timeout) FROM event_errors WHERE event_time > '2022-08-01'", settings=settings)

import clickhouse_connect

client = clickhouse_connect.get_client()

# 테이블 생성
result = client.command("CREATE TABLE test_command (col_1 String, col_2 DateTime) ENGINE MergeTree ORDER BY tuple()")
print(result)  # query_id가 포함된 QuerySummary 반환

# 테이블 정의 조회
result = client.command("SHOW CREATE TABLE test_command")
print(result)
# 출력:
# CREATE TABLE default.test_command
# (
#     `col_1` String,
#     `col_2` DateTime
# )
# ENGINE = MergeTree
# ORDER BY tuple()

# 테이블 삭제
client.command("DROP TABLE test_command")

import clickhouse_connect

client = clickhouse_connect.get_client()

# 단일 값 결과
count = client.command("SELECT count() FROM system.tables")
print(count)
# 출력: 151

# 서버 버전
version = client.command("SELECT version()")
print(version)
# 출력: "25.8.2.29"

import clickhouse_connect

client = clickhouse_connect.get_client()

# 클라이언트 측 매개변수 사용
table_name = "system"
result = client.command(
    "SELECT count() FROM system.tables WHERE database = %(db)s",
    parameters={"db": table_name}
)

# 서버 측 매개변수 사용
result = client.command(
    "SELECT count() FROM system.tables WHERE database = {db:String}",
    parameters={"db": "system"}
)

import clickhouse_connect

client = clickhouse_connect.get_client()

# 특정 설정을 사용해 명령 실행
result = client.command(
    "OPTIMIZE TABLE large_table FINAL",
    settings={"optimize_throw_if_noop": 1}
)

import clickhouse_connect

client = clickhouse_connect.get_client()

# 간단한 SELECT 쿼리
result = client.query("SELECT name, database FROM system.tables LIMIT 3")

# 결과를 행 단위로 확인
for row in result.result_rows:
    print(row)
# 출력:
# ('CHARACTER_SETS', 'INFORMATION_SCHEMA')
# ('COLLATIONS', 'INFORMATION_SCHEMA')
# ('COLUMNS', 'INFORMATION_SCHEMA')

# 컬럼 이름과 타입 확인
print(result.column_names)
# 출력: ("name", "database")
print([col_type.name for col_type in result.column_types])
# 출력: ['String', 'String']

import clickhouse_connect

client = clickhouse_connect.get_client()

result = client.query("SELECT number, toString(number) AS str FROM system.numbers LIMIT 3")

# 행 지향 접근 (기본값)
print(result.result_rows)
# 출력: [[0, "0"], [1, "1"], [2, "2"]]

# 컬럼 지향 접근
print(result.result_columns)
# 출력: [[0, 1, 2], ["0", "1", "2"]]

# 이름 있는 결과 (딕셔너리 목록)
for row_dict in result.named_results():
    print(row_dict)
# 출력: 
# {"number": 0, "str": "0"}
# {"number": 1, "str": "1"}
# {"number": 2, "str": "2"}

# 첫 번째 행을 딕셔너리로
print(result.first_item)
# 출력: {"number": 0, "str": "0"}

# 첫 번째 행을 튜플로
print(result.first_row)
# 출력: (0, "0")

import clickhouse_connect

client = clickhouse_connect.get_client()

# 딕셔너리 매개변수 사용 (printf 스타일)
query = "SELECT * FROM system.tables WHERE database = %(db)s AND name LIKE %(pattern)s"
parameters = {"db": "system", "pattern": "%query%"}
result = client.query(query, parameters=parameters)

# 튜플 매개변수 사용
query = "SELECT * FROM system.tables WHERE database = %s LIMIT %s"
parameters = ("system", 5)
result = client.query(query, parameters=parameters)

import clickhouse_connect

client = clickhouse_connect.get_client()

# 서버 측 바인딩 (보안이 강화되고 SELECT 쿼리 성능이 향상됨)
query = "SELECT * FROM system.tables WHERE database = {db:String} AND name = {tbl:String}"
parameters = {"db": "system", "tbl": "query_log"}

result = client.query(query, parameters=parameters)

import clickhouse_connect

client = clickhouse_connect.get_client()

# 쿼리와 함께 ClickHouse 설정을 지정합니다
result = client.query(
    "SELECT sum(number) FROM numbers(1000000)",
    settings={
        "max_block_size": 100000,
        "max_execution_time": 30
    }
)

import clickhouse_connect

client = clickhouse_connect.get_client()

# 행 지향 데이터: 각 내부 리스트가 하나의 행
data = [
    [1, "Alice", 25],
    [2, "Bob", 30],
    [3, "Joe", 28],
]

client.insert("users", data, column_names=["id", "name", "age"])

import clickhouse_connect

client = clickhouse_connect.get_client()

# 컬럼 지향 데이터: 각 내부 리스트는 하나의 컬럼
data = [
    [1, 2, 3],  # id 컬럼
    ["Alice", "Bob", "Joe"],  # name 컬럼
    [25, 30, 28],  # age 컬럼
]

client.insert("users", data, column_names=["id", "name", "age"], column_oriented=True)

import clickhouse_connect

client = clickhouse_connect.get_client()

# 서버에 DESCRIBE 쿼리를 보내지 않으려 할 때 유용합니다
data = [
    [1, "Alice", 25],
    [2, "Bob", 30],
    [3, "Joe", 28],
]

client.insert(
    "users",
    data,
    column_names=["id", "name", "age"],
    column_type_names=["UInt32", "String", "UInt8"],
)

import clickhouse_connect

client = clickhouse_connect.get_client()

data = [
    [1, "Alice", 25],
    [2, "Bob", 30],
]

# 특정 데이터베이스의 테이블에 삽입
client.insert(
    "users",
    data,
    column_names=["id", "name", "age"],
    database="production",
)