> ## Documentation Index > Fetch the complete documentation index at: https://private-7c7dfe99-vortex-format.mintlify.site/llms.txt > Use this file to discover all available pages before exploring further. > 기존 Airflow DAG를 커뮤니티 airflow-clickhouse-plugin에서 공식 apache-airflow-providers-clickhousedb provider로 마이그레이션합니다 # airflow-clickhouse-plugin에서 ClickHouse provider로 마이그레이션하기 [공식 ClickHouse provider](/ko/integrations/connectors/data-ingestion/etl-tools/airflow-and-clickhouse)가 등장하기 전에는 대부분의 Airflow 사용자가 커뮤니티 패키지 [airflow-clickhouse-plugin](https://github.com/bryzgaloff/airflow-clickhouse-plugin)을 통해 ClickHouse에 연결했습니다. 이 가이드에서는 기존 배포를 `apache-airflow-providers-clickhousedb`로 마이그레이션하는 방법을 설명합니다. provider 자체의 동작 방식은 [Apache Airflow를 ClickHouse에 연결하기](/ko/integrations/connectors/data-ingestion/etl-tools/airflow-and-clickhouse)를 참조하십시오.

왜 네이티브 provider인가?

provider는 Airflow가 서드파티 시스템과 통합되는 방식입니다. provider는 Airflow 생태계의 나머지 구성 요소와 함께 릴리스되고 테스트되며 문서화되며, 이 provider는 Airflow 커뮤니티가 ClickHouse 팀과 함께 유지 관리합니다. 또한 커뮤니티가 유지 관리하는 driver가 아니라, ClickHouse가 직접 개발하고 지원하는 Python client인 [clickhouse-connect](/ko/integrations/language-clients/python/index)를 기반으로 구축되었습니다. 따라서 새로운 서버 기능과 수정 사항이 공식 지원 경로를 통해 Airflow 사용자에게 전달됩니다. provider로 전환하면 공식적으로 관리되는 package와 표준 `common.sql` operator 및 sensor, 그리고 다른 모든 데이터베이스와 마찬가지로 Airflow UI에 표시되는 connection 유형을 사용할 수 있습니다. 두 package의 차이는 import 경로에 그치지 않습니다. plugin은 `clickhouse-driver`를 사용해 **네이티브 TCP 프로토콜**로 ClickHouse와 통신합니다. provider는 `clickhouse-connect`를 사용해 \*\*HTTP(S)\*\*로 통신하며, ClickHouse 전용 operator를 별도로 제공하는 대신 범용 `common.sql` operator에 연결됩니다. 무엇이든 변경하기 전에 가이드 전체를 한 번 읽어 보십시오. 특히 connection 변경은 모든 DAG에 동시에 영향을 미칩니다.

한눈에 보기

| | `airflow-clickhouse-plugin` | `apache-airflow-providers-clickhousedb` | | - | - | - | | 가져오기 루트 | `airflow_clickhouse_plugin` | `airflow.providers.clickhousedb` | | 드라이버 | `clickhouse-driver` | `clickhouse-connect` | | 프로토콜 / 기본 포트 | 네이티브 TCP, `9000` (TLS 사용 시 `9440`) | HTTP, `8123` (TLS 사용 시 `8443`) | | 연결 유형 | 등록된 유형 없음, 모든 유형 사용 가능 | `clickhouse` | | 연결 추가 옵션 | `clickhouse_driver.Client`에 그대로 전달 | 정해진 키 집합, [추가 연결 옵션](/ko/integrations/connectors/data-ingestion/etl-tools/airflow-and-clickhouse#extra-connection-options) 참조 | | 연산자 및 센서 | `ClickHouseOperator`, `ClickHouseSensor` 및 `ClickHouse` 접두사가 붙은 `common.sql` 래퍼 | `common.sql` 연산자와 센서를 직접 사용 | | 후크 | `ClickHouseHook`(`BaseHook`) 및 `ClickHouseDbApiHook`(`DbApiHook`) | `ClickHouseHook`(`DbApiHook`) 하나 | | 최소 Airflow 버전 | 2.0 | 2.11 |

1단계: 사전 요구 사항 확인 및 설치

이 provider를 사용하려면 Airflow 2.11 이상과 `apache-airflow-providers-common-sql` 1.32.0 이상이 필요합니다. 그보다 이전 버전을 사용하고 있다면 먼저 Airflow를 업그레이드하십시오. ```bash theme={null} pip install apache-airflow-providers-clickhousedb ``` 두 package는 서로 다른 Python 네임스페이스에 존재하므로, DAG를 하나씩 마이그레이션하는 동안 두 package를 함께 설치해 둘 수 있습니다. 더 이상 아무 곳에서도 plugin을 가져오지 않는다면 이를 제거하십시오: ```bash theme={null} pip uninstall airflow-clickhouse-plugin clickhouse-driver ```

2단계: 연결 업데이트

이 단계를 건너뛰면 문제가 발생합니다. 기존 ClickHouse 연결은 모두 네이티브 포트를 가리키고 있지만, provider는 HTTP 포트가 필요합니다. | 필드 | Plugin | Provider | | - | - | - | | 연결 유형 | 모든 유형(주로 `sqlite` 또는 `generic`) | `clickhouse` | | 포트, 일반 | `9000` | `8123` | | 포트, TLS | `9440` | `8443` | | Login | driver 기본값 `default` | 동일 | | Schema | 데이터베이스 | 동일 | ClickHouse가 firewall 또는 load balancer 뒤에 있다면, 전환하기 전에 worker에서 HTTP 포트에 연결할 수 있는지 확인하십시오. 또한 서버에서 HTTP 인터페이스가 활성화되어 있는지(서버 구성의 `http_port` 또는 `https_port`) 확인하십시오. [ClickHouse Cloud](/ko/products/cloud/getting-started/intro)는 `8443` 포트에서만 HTTPS를 제공합니다. URI 형태로 저장된 연결(`clickhouse://user:pass@host:9000/db?secure=true`)은 Airflow가 URI scheme에서 연결 유형을 가져오므로 이미 `clickhouse` 연결 유형으로 설정되어 있습니다. 이 경우 포트와 extra keys만 변경하면 됩니다. 쿼리 문자열 값은 JSON으로 파싱되므로 `secure=true`는 Boolean 값으로 유지됩니다.

연결 extra 옵션

플러그인은 `extra`의 모든 키를 `clickhouse_driver.Client`에 그대로 전달했기 때문에, 연결에는 어떤 `clickhouse-driver` keyword argument든 포함될 수 있었습니다. provider는 고정된 키 집합만 읽고, 나머지는 `client_kwargs`를 통해 전달합니다. 다음과 같이 변환하십시오: | 플러그인 extra (`clickhouse-driver`) | Provider extra | 참고 | | - | - | - | | `secure` | `secure` | 변경 없음. 포트도 함께 변경해야 합니다. | | `verify` | `verify` | 변경 없음. | | `settings` | `session_settings` | 내용은 동일하며 키만 변경되었습니다. | | `compression` | `compress` | Boolean. 알고리즘 이름은 driver마다 다르므로 `true`가 안전한 선택입니다. | | `connect_timeout` | `connect_timeout` | 변경 없음. | | `send_receive_timeout` | `send_receive_timeout` | 변경 없음. | | `client_name` | `client_name` | 의미가 달라집니다. provider는 항상 `apache-airflow/ apache-airflow-providers-clickhousedb/`을 전송하며, 지정한 값은 레이블로 뒤에 덧붙입니다. | | `ca_certs` | `client_kwargs.ca_cert` | CA bundle 경로. | | `certfile` / `keyfile` | `client_kwargs.client_cert` / `client_kwargs.client_cert_key` | Mutual TLS. | | `server_hostname` | `client_kwargs.server_host_name` | TLS SNI override. | | `alt_hosts`, `round_robin` | 대응 항목 없음 | 클라이언트는 연결에 지정된 단일 host에만 접속합니다. failover 목록에 의존해 왔다면, 연결 대상을 load balancer 또는 ClickHouse Cloud 엔드포인트로 지정하십시오. | | `sync_request_timeout`, `tcp_keepalive`, `compress_block_size` | 삭제 | native protocol 전용입니다. | | `ssl_version`, `ciphers`, `use_numpy`, `client_revision`, `settings_is_important`, `opentelemetry_traceparent`, `opentelemetry_tracestate` | 삭제 | `clickhouse-connect`에는 대응 항목이 없습니다. | 변경 전: ```json theme={null} { "conn_type": "sqlite", "host": "ch.example.com", "port": 9440, "login": "airflow", "password": "secret", "schema": "analytics", "extra": { "secure": true, "settings": {"max_execution_time": 300}, "compression": true } } ``` 변경 후: ```json theme={null} { "conn_type": "clickhouse", "host": "ch.example.com", "port": 8443, "login": "airflow", "password": "secret", "schema": "analytics", "extra": { "secure": true, "session_settings": {"max_execution_time": 300}, "compress": true } } ``` DAG를 수정하기 전에 마이그레이션된 각 connection을 검증하십시오: ```bash theme={null} airflow connections test clickhouse_default ```

3단계: import 교체

| 플러그인 클래스 | provider 대체 대상 | | - | - | | `airflow_clickhouse_plugin.hooks.clickhouse.ClickHouseHook` | `airflow.providers.clickhousedb.hooks.clickhouse.ClickHouseHook` | | `airflow_clickhouse_plugin.hooks.clickhouse_dbapi.ClickHouseDbApiHook` | `airflow.providers.clickhousedb.hooks.clickhouse.ClickHouseHook` | | `airflow_clickhouse_plugin.operators.clickhouse.ClickHouseOperator` | `airflow.providers.common.sql.operators.sql.SQLExecuteQueryOperator` | | `airflow_clickhouse_plugin.sensors.clickhouse.ClickHouseSensor` | `airflow.providers.common.sql.sensors.sql.SqlSensor` | | `airflow_clickhouse_plugin.operators.clickhouse_dbapi.ClickHouseSQLExecuteQueryOperator` | `airflow.providers.common.sql.operators.sql.SQLExecuteQueryOperator` | | `...clickhouse_dbapi.ClickHouseSQLCheckOperator` | `...common.sql.operators.sql.SQLCheckOperator` | | `...clickhouse_dbapi.ClickHouseSQLValueCheckOperator` | `...common.sql.operators.sql.SQLValueCheckOperator` | | `...clickhouse_dbapi.ClickHouseSQLIntervalCheckOperator` | `...common.sql.operators.sql.SQLIntervalCheckOperator` | | `...clickhouse_dbapi.ClickHouseSQLThresholdCheckOperator` | `...common.sql.operators.sql.SQLThresholdCheckOperator` | | `...clickhouse_dbapi.ClickHouseSQLColumnCheckOperator` | `...common.sql.operators.sql.SQLColumnCheckOperator` | | `...clickhouse_dbapi.ClickHouseSQLTableCheckOperator` | `...common.sql.operators.sql.SQLTableCheckOperator` | | `...clickhouse_dbapi.ClickHouseBranchSQLOperator` | `...common.sql.operators.sql.BranchSQLOperator` | | `airflow_clickhouse_plugin.sensors.clickhouse_dbapi.ClickHouseSqlSensor` | `airflow.providers.common.sql.sensors.sql.SqlSensor` | `ClickHouse` 접두사가 붙은 `common.sql` 래퍼는 플러그인의 후크를 주입하기 위한 용도로만 존재했습니다. provider가 `clickhouse` connection 유형을 등록하기 때문에, 접두사가 없는 `common.sql` 클래스는 connection에서 후크를 직접 확인합니다. 이러한 래퍼를 사용했다면, 마이그레이션은 대개 import 줄을 수정하고 `ClickHouse` 접두사를 제거한 뒤 `conn_id`를 명시적으로 전달하는 것으로 끝납니다([7단계](#step-7-the-commonsql-wrapper-family) 참조).

4단계: `ClickHouseOperator`에서 `SQLExecuteQueryOperator`로

| `ClickHouseOperator` 인수 | `SQLExecuteQueryOperator` 대응 항목 | 참고 | | - | - | - | | `sql` | `sql` | 문자열, 문자열 목록 또는 `.sql` 파일 경로입니다. 양쪽 모두 템플릿이 적용됩니다. | | `clickhouse_conn_id` | `conn_id` | plugin의 기본값은 `clickhouse_default`였습니다. `SQLExecuteQueryOperator`에는 기본값이 없으므로 모든 작업에 `conn_id`를 전달하거나 `default_args`에 한 번 설정하십시오. | | `database` | `database` | 변경 없음. | | `SELECT`용 `parameters` | `parameters` | `%(name)s` 플레이스홀더는 그대로 동작합니다. `{name:Type}` 형식의 서버 측 binding도 사용할 수 있습니다. | | `INSERT`용 `parameters` (행 목록) | `SQLInsertRowsOperator` | 행(또는 XCom)을 `rows`로 전달하고, `columns`로 컬럼 이름을 지정한 뒤, `clickhouse-connect`가 네이티브 insert를 사용하도록 `insert_args={"executemany": True}`를 설정하십시오. 또는 `@task`에서 `ClickHouseHook.bulk_insert_rows`를 호출하는 방법도 있습니다. [5단계](#step-5-clickhousehookexecute-to-dbapihook-methods)를 참조하십시오. | | `settings` | `hook_params={"session_settings": {...}}` | `hook_params`를 통해 템플릿이 적용됩니다. 값은 connection의 `session_settings` 위에 병합됩니다. | | `do_xcom_push` | `do_xcom_push` | 플래그는 동일하지만, 다중 SQL 문 작업에서는 결과 형태가 다릅니다. plugin은 마지막 SQL 문의 결과만 푸시했습니다. SQL 문 목록을 사용하면 provider는 각 SQL 문마다 항목이 하나씩 담긴 목록을 푸시합니다. [다중 SQL 문 결과](#multi-statement-results)를 참조하십시오. | | `query_id` | `hook_params={"session_settings": {"query_id": "..."}}` | `hook_params`를 통해 템플릿이 적용됩니다. plugin과 마찬가지로, 다중 SQL 문 작업의 모든 SQL 문에 동일한 id가 전송됩니다. 설정하지 않으면 `clickhouse-connect`가 SQL 문마다 고유한 id를 생성하며, `handler`에서 `cursor.summary[-1]["query_id"]`로 마지막 id를 읽을 수 있습니다. | | `with_column_types` | `handler` | 행과 함께 `cursor.description`을 읽는 handler를 전달하십시오. [컬럼 타입 유지](#keeping-column-types)를 참조하십시오. | | `external_tables` | `ClickHouseHook.get_client()` | `clickhouse_connect.driver.external.ExternalData`를 `client.query`와 함께 사용하십시오. | | `columnar` | `ClickHouseHook.get_client()` | `client.query(...).result_columns`. | | `types_check` | 제거 | native protocol 전용입니다. | 이전: ```python theme={null} from airflow_clickhouse_plugin.operators.clickhouse import ClickHouseOperator update_income_aggregate = ClickHouseOperator( task_id="update_income_aggregate", clickhouse_conn_id="clickhouse_test", database="default", sql=( """ INSERT INTO aggregate SELECT eventDt, sum(price * qty) AS income FROM sales WHERE eventDt = '{{ ds }}' GROUP BY eventDt """, """ SELECT sum(income) FROM aggregate WHERE eventDt BETWEEN '{{ data_interval_start | ds }}' AND '{{ data_interval_end | ds }}' """, ), settings={"max_execution_time": 600}, ) ``` 변경 후: ```python theme={null} from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator update_income_aggregate = SQLExecuteQueryOperator( task_id="update_income_aggregate", conn_id="clickhouse_test", database="default", sql=[ """ INSERT INTO aggregate SELECT eventDt, sum(price * qty) AS income FROM sales WHERE eventDt = '{{ ds }}' GROUP BY eventDt """, """ SELECT sum(income) FROM aggregate WHERE eventDt BETWEEN '{{ data_interval_start | ds }}' AND '{{ data_interval_end | ds }}' """, ], hook_params={"session_settings": {"max_execution_time": 600}}, ) ```

다중 SQL 문 결과

plugin은 **마지막** SQL 문의 결과를 XCom에 푸시했습니다. `SQLExecuteQueryOperator`는 `sql`이 리스트일 때 SQL 문마다 하나씩 결과를 반환하므로, plugin이 `[(12345.0,)]`를 푸시했던 것과 달리 위 예시는 `[[], [(12345.0,)]]`를 푸시합니다. 다음 중 하나를 선택하십시오. * 다운스트림의 `xcom_pull`이 마지막 원소를 가져오도록 변경합니다. * SQL 문들을 `;`로 구분한 단일 문자열로 전달하고 `split_statements=True`로 설정합니다. 기본값인 `return_last=True`와 함께 사용하면 연산자가 마지막 SQL 문의 행만 푸시하므로 plugin과 동일하게 동작합니다. ```python theme={null} SQLExecuteQueryOperator( task_id="update_income_aggregate", conn_id="clickhouse_test", sql=""" INSERT INTO aggregate SELECT eventDt, sum(price * qty) AS income FROM sales WHERE eventDt = '{{ ds }}' GROUP BY eventDt; SELECT sum(income) FROM aggregate WHERE eventDt = '{{ ds }}' """, split_statements=True, ) ``` `parameters`를 통해 행 목록을 삽입하던 `ClickHouseOperator`는 `SQLInsertRowsOperator`로 대체됩니다: ```python theme={null} from airflow.providers.common.sql.operators.sql import SQLInsertRowsOperator load_rows = SQLInsertRowsOperator( task_id="load_rows", conn_id="clickhouse_default", table_name="some_ch_table", columns=["id", "name"], rows=extract_task.output, insert_args={"executemany": True}, ) ``` 항상 `columns`를 전달하십시오. 전달하지 않으면 연산자가 SQLAlchemy를 통해 테이블을 조회합니다.

컬럼 타입 유지하기

`with_column_types=True`는 `(rows, [(name, type), ...])`를 반환했습니다. handler를 사용하면 이를 재현할 수 있으며, `clickhouse-connect` cursor는 `cursor.description`에 ClickHouse 타입 이름을 제공합니다: ```python theme={null} def fetch_with_column_types(cursor): return cursor.fetchall(), [(col[0], col[1]) for col in cursor.description] SQLExecuteQueryOperator( task_id="typed_query", conn_id="clickhouse_default", sql="SELECT id, name FROM users LIMIT 10", handler=fetch_with_column_types, ) ```

Step 5: `ClickHouseHook.execute`에서 `DbApiHook` 메서드로

플러그인의 후크는 `clickhouse_driver.Client.execute`를 그대로 반영한 `execute` 메서드 하나만 노출했습니다. provider의 후크는 `DbApiHook`이므로, 다른 모든 SQL provider와 동일한 표준 메서드를 사용할 수 있습니다: `run`, `get_records`, `get_first`, `get_pandas_df`, `get_df`, `insert_rows`, `test_connection`. `clickhouse_conn_id` 및 `database` 생성자 인수는 변경되지 않았습니다. | 플러그인 | Provider | | - | - | | `hook.execute("SELECT ...")` | `hook.get_records("SELECT ...")` | | `hook.execute("SELECT ...", params={"d": ds})` | `hook.get_records("SELECT ...", parameters={"d": ds})` | | `hook.execute("SELECT count() ...")[0][0]` | `hook.get_first("SELECT count() ...")[0]` | | `hook.execute("INSERT INTO t VALUES", rows)` | `hook.bulk_insert_rows("t", rows, column_names=[...])` | | `hook.execute(["SET ...", "INSERT ...", "SELECT ..."])` | `hook.run([...], handler=fetch_all_handler, return_last=True)` | | `hook.execute(..., settings={...})` | `ClickHouseHook(session_settings={...})` | | `hook.execute(..., external_tables=..., columnar=..., query_id=...)` | `hook.get_client().query(...)` | | `clickhouse_driver.Client`를 반환하는 `hook.get_conn()` | `clickhouse_connect` `Client`를 반환하는 `hook.get_client()` | `fetch_all_handler` 및 그 외 handler는 `airflow.providers.common.sql.hooks.handlers`에서 가져옵니다. 플러그인 시절 코드에서 가장 흔한 후크 사용 패턴은 bulk insert입니다. DB-API 커서가 행을 SQL 문자열에 포맷해 넣으려 하기 때문에, 단순한 `run` 호출로는 처리할 수 없습니다. 대신 native insert를 사용하십시오: 이전: ```python theme={null} from airflow_clickhouse_plugin.hooks.clickhouse import ClickHouseHook def sqlite_to_clickhouse(): records = SqliteHook().get_records("SELECT id, name FROM some_sqlite_table") ClickHouseHook().execute("INSERT INTO some_ch_table VALUES", records) ``` 변경 후: ```python theme={null} from airflow.providers.clickhousedb.hooks.clickhouse import ClickHouseHook def sqlite_to_clickhouse(): records = SqliteHook().get_records("SELECT id, name FROM some_sqlite_table") ClickHouseHook().bulk_insert_rows( "some_ch_table", records, column_names=["id", "name"], batch_size=100_000 ) ``` `bulk_insert_rows`에는 `column_names`가 필요합니다. `batch_size`는 선택 사항이며, 매우 큰 입력에서 메모리 사용량을 제한하는 역할을 합니다. 범용 메서드인 `insert_rows(table, rows, target_fields=[...], executemany=True)`도 최종적으로 네이티브 삽입으로 처리되지만, `executemany=True`일 때만 그렇고 기본값에서는 행마다 HTTP request를 한 번씩 전송합니다. DB-API 계층에서 다루지 못하는 작업은 `get_client()`를 사용하면 되며, 이 메서드는 Airflow connection 정보로 구성된 raw `clickhouse-connect` 클라이언트를 반환합니다. plugin이 노출했던 모든 `clickhouse-driver` 전용 인수를 대체하는 수단입니다: ```python theme={null} from clickhouse_connect.driver.external import ExternalData hook = ClickHouseHook() with hook.get_client() as client: ext = ExternalData(file_name="ids", structure="id UInt64", data=b"1\n2\n3\n") result = client.query( "SELECT * FROM events WHERE id IN ids", external_data=ext, settings={"query_id": "my-traceable-id"}, ) columns = result.result_columns ```

Step 6: `ClickHouseSensor`에서 `SqlSensor`로

호출 가능 객체(callable)의 입력이 바뀌는 유일한 대체 작업입니다. | | `ClickHouseSensor` | `SqlSensor` | | - | - | - | | 성공 호출 가능 객체 | `is_success(result)` | `success(cell)` | | 실패 호출 가능 객체 | `is_failure(result)` | `failure(cell)` | | 호출 가능 객체가 받는 값 | 마지막 SQL 문의 전체 결과, 즉 행 튜플의 `list` | 기본적으로 첫 번째 행의 첫 번째 컬럼(`selector=itemgetter(0)`) | | 기본 성공 조건 | `bool(result)`, 행이 하나라도 반환되면 true | 행이 반환된 경우 `bool(first cell)`, 반환된 행이 없으면 `False` | | `sql` | 문자열 또는 SQL 문 목록, 그리고 모든 `ClickHouseOperator` 인수 | 단일 문자열. 데이터베이스 및 세션 설정에는 `hook_params`를 사용하십시오. | | 연결 ID | `clickhouse_conn_id`, 기본값 `clickhouse_default` | `conn_id`, 필수 | | 반환된 행이 없을 때 | `is_success([])`, 기본값으로는 `False` | `False`, 또는 `fail_on_empty=True`인 경우 오류 | plugin이 전체 결과 집합을 그대로 넘겨주었기 때문에, 기존에 동작하던 센서 코드는 결과에 인덱스로 접근합니다. 마이그레이션할 때는 이러한 인덱싱을 제거하십시오. 변경 전: ```python theme={null} from airflow_clickhouse_plugin.sensors.clickhouse import ClickHouseSensor ClickHouseSensor( task_id="poke_events_count", database="monitor", sql="SELECT count() FROM warnings WHERE eventDate = '{{ ds }}'", is_success=lambda result: result[0][0] > 10000, ) ``` 변경 후: ```python theme={null} from airflow.providers.common.sql.sensors.sql import SqlSensor SqlSensor( task_id="poke_events_count", conn_id="clickhouse_default", hook_params={"database": "monitor"}, sql="SELECT count() FROM warnings WHERE eventDate = '{{ ds }}'", success=lambda cnt: cnt > 10000, ) ``` 호출 가능 객체에서 행 전체가 필요하다면 `selector=lambda row: row`를 전달하십시오. 모든 행이 필요하다면, 쿼리가 단일 불리언 값이나 개수를 반환하도록 검사 로직을 SQL로 작성하십시오.

7단계: `common.sql` 래퍼 계열

`ClickHouseSQLExecuteQueryOperator`, `ClickHouseSqlSensor` 및 그 외 `ClickHouse` 접두사가 붙은 래퍼를 사용하던 코드는 손볼 부분이 가장 적습니다: * import를 `common.sql` 모듈로 변경하고 클래스 이름에서 `ClickHouse` 접두사를 제거하십시오. * `conn_id`를 명시적으로 전달하십시오. 래퍼는 `conn_id`가 없거나 `None`이면 `clickhouse_default`로 처리했지만, `common.sql` 클래스에는 기본값이 없어 지정하지 않으면 실패합니다. `default_args={"conn_id": "clickhouse_default"}`를 사용하면 DAG 전체에 적용됩니다. * 연산자의 `database=`와 센서의 `hook_params={"schema": ...}`는 계속 동작합니다. provider의 후크는 `schema`를 `database`의 별칭으로 취급합니다. * `ClickHouseDbApiHook`은 `ClickHouseHook`으로 바뀝니다. 생성자 인수 `schema`는 여전히 별칭으로 허용되며, ClickHouse의 네이티브 표기는 `database`이므로 둘 다 지정된 경우 `database`가 우선합니다. * connection에는 2단계에서 설명한 포트 및 extras 변경이 여전히 필요합니다. 래퍼 역시 native protocol을 사용했기 때문입니다.

검토해야 할 동작 차이

코드가 정상적으로 컴파일된 뒤에도 런타임 동작이 몇 가지 달라집니다. **INSERT 작업의 XCom 값.** 플러그인은 `clickhouse-driver`가 반환한 값을 그대로 푸시했으며, 파라미터를 사용한 `VALUES` 삽입에서는 이 값이 삽입된 행 수였습니다. 반면 provider는 행을 반환하지 않는 SQL 문에 대해 빈 결과 집합을 푸시합니다. 따라서 XCom에서 행 수를 읽던 다운스트림 작업은 후속 `SELECT count()` 등 다른 방법으로 행 수를 가져와야 합니다. **세션 설정과 `SET` 문.** 두 패키지 모두 단일 연결에서 여러 SQL 문 목록을 실행하며, `clickhouse-connect`는 기본적으로 클라이언트당 세션을 생성하므로 목록 앞부분의 `SET` 문은 이후 SQL 문에도 계속 적용됩니다. 그래도 `session_settings`를 사용하는 것이 좋습니다. 명시적이고 템플릿을 적용할 수 있으며, 서버나 중간 프록시가 세션을 유지하는지 여부와 무관하게 동일하게 동작합니다. DAG가 `SET`에 의존한다면 해당 환경에서 동작을 직접 확인하십시오. **예외.** 오류는 이제 `clickhouse_driver.errors.ServerException` 및 `NetworkError`가 아니라 `clickhouse_connect.driver.exceptions.DatabaseError`, `OperationalError` 또는 `ProgrammingError`로 발생합니다. 예외 타입을 검사하는 `except` 절, `on_failure_callback` 코드, 재시도 로직을 업데이트하십시오. **압축.** `clickhouse-driver`는 `compression`을 설정하지 않으면 압축을 사용하지 않았지만, `clickhouse-connect`는 기본적으로 HTTP 응답 압축을 활성화하고 서버와 알고리즘을 협상합니다. 이전 동작으로 되돌리려면 연결 extra에 `"compress": false`를 설정하십시오. 네이티브 압축을 위해 플러그인에 필요했던 `clickhouse-cityhash` 패키지는 더 이상 필요하지 않으며, `lz4`는 `clickhouse-connect` 의존성으로 계속 설치됩니다. **타입 매핑.** 두 driver 모두 네이티브 Python 타입을 반환하지만, 서로 코드 베이스가 다릅니다. 시간대가 지정된 `DateTime64`, `Decimal`, `UUID`, `Nullable` 컬럼, 중첩된 `Array` 또는 `Map` 값의 정확한 타입에 의존하는 작업을 검토하십시오. 특히 결과가 XCom에 푸시되어 다운스트림에서 사용되는 경우 주의가 필요합니다. **`system.query_log`에서의 쿼리 식별.** 쿼리는 이제 HTTP 인터페이스를 통해 전달되므로 `interface = 1`이 아닌 `interface = 2`로 표시되며, [`system.query_log`](/ko/reference/system-tables/query_log)의 `http_user_agent` 컬럼에는 Airflow 및 provider 버전과, 설정된 경우 `client_name` extra가 함께 담깁니다. native protocol이나 `clickhouse-driver` 클라이언트 이름을 기준으로 필터링하던 모니터링은 업데이트해야 합니다. 행을 반환하지 않는 `SELECT`는 항목을 하나 더 생성합니다. DB-API 커서가 컬럼 메타데이터를 얻기 위해 `SELECT * FROM (...) LIMIT 0`을 실행하기 때문입니다. **HTTP를 통한 타임아웃.** `send_receive_timeout`은 이제 HTTP 읽기 타임아웃이며, worker와 ClickHouse 사이에 있는 프록시나 로드 밸런서는 요청에 각자의 idle timeout을 적용합니다. native protocol에서 수 분 동안 실행되던 SQL 문은 이러한 제한값을 높여야 할 수 있습니다. **연결 처리.** 후크는 `run` 또는 `get_client` 호출마다 `clickhouse-connect` 클라이언트를 생성하며, 이는 플러그인이 `execute`마다 새 native connection을 열었던 방식과 같습니다. 클라이언트들은 프로세스 전역 HTTP 연결 풀을 공유하므로, `get_client()`로 얻은 클라이언트에 `close()`를 호출하는 것은 필수는 아니지만 권장되는 습관입니다. 풀은 작업 프로세스가 종료될 때 해제됩니다. 클라이언트는 context manager이므로 `with hook.get_client() as client:` 형태가 가장 깔끔합니다.

체크리스트

1. Airflow 버전이 2.11 이상입니다. 2. worker에서 HTTP 포트에 연결할 수 있으며, HTTP 엔드포인트용 TLS 인증서가 유효합니다. 3. 모든 ClickHouse connection: 타입은 `clickhouse`, 포트는 `8123` 또는 `8443`, extras는 2단계에 따라 변환했으며, `airflow connections test`로 검증했습니다. 4. 3단계에 따라 import를 대체했으며, `common.sql` 래퍼에서 `ClickHouse` 프리픽스를 제거했습니다. 5. 연산자와 센서에서 `clickhouse_conn_id`를 `conn_id`로 이름을 변경했고, 플러그인의 기본값에 의존하던 모든 작업에 `conn_id`를 설정했습니다. 6. `settings=`를 `hook_params={"session_settings": ...}` 안으로 옮겼습니다. 7. `hook.execute("INSERT ... VALUES", rows)`를 `bulk_insert_rows`로 대체했고, `ClickHouseOperator(parameters=rows)`를 `SQLInsertRowsOperator`로 대체했습니다. 8. 센서 호출 가능 객체을 `result[0][0]` 대신 cell 값 자체를 사용하도록 수정했습니다. 9. `with_column_types`, `external_tables`, `columnar`, `query_id`, `types_check`를 사용하던 부분을 `handler` 또는 `get_client()`로 재작성했습니다. 10. 다중 SQL 문 및 INSERT 작업의 XCom을 사용하는 다운스트림 소비자를 검토했습니다. 11. `clickhouse_driver` 예외를 처리하던 코드를 업데이트했습니다. 12. `airflow-clickhouse-plugin`과 `clickhouse-driver`를 제거했습니다.

AI 코딩 어시스턴트 활용

위의 매핑은 코딩 어시스턴트가 DAG 리포지토리에 그대로 적용할 수 있도록 의도적으로 기계적으로 구성했습니다. 다음 프롬프트가 효과적이었습니다: ```text theme={null} Migrate this repository from airflow-clickhouse-plugin to apache-airflow-providers-clickhousedb following https://clickhouse.com/docs/integrations/airflow/migrating-from-airflow-clickhouse-plugin - Replace every airflow_clickhouse_plugin import per the class mapping table. - Rename clickhouse_conn_id to conn_id on operators and sensors, not on hooks; add conn_id="clickhouse_default" wherever a task relied on the plugin's default. - Move settings= into hook_params={"session_settings": ...}. - Replace hook.execute(...) with get_records, get_first, run or bulk_insert_rows according to the hook table; never pass a list of rows as parameters. Replace ClickHouseOperator(parameters=) with SQLInsertRowsOperator. - Rewrite sensor callables to receive the first cell instead of the full result. - Flag, but do not silently rewrite, any use of with_column_types, external_tables, columnar, query_id or types_check, and any XCom consumer of an INSERT task. - List every Airflow connection that must change port and extras; do not edit connections yourself. Show the diff and a summary of items flagged for human review. ``` diff를 검토하십시오. connection 변경, 센서 의미론, 그리고 XCom consumer는 자동화된 재작성이 잘못되기 쉬운 지점입니다.

관련 콘텐츠

* [Apache Airflow를 ClickHouse에 연결하기](/ko/integrations/connectors/data-ingestion/etl-tools/airflow-and-clickhouse) * [`clickhouse-connect` Python 클라이언트](/ko/integrations/language-clients/python/index) * [`apache-airflow-providers-clickhousedb` 참고 문서](https://airflow.apache.org/docs/apache-airflow-providers-clickhousedb/) * [GitHub의 airflow-clickhouse-plugin](https://github.com/bryzgaloff/airflow-clickhouse-plugin)