из interview.utils импортировать get_clickhouse_client

Условие

из interview.utils импортировать get_clickhouse_client из airflow импортировать DAG из airflow.operators.python импортировать PythonOperator из airflow.sensors.external_task импортировать ExternalTaskSensor из datetime импортировать datetime импортировать pandas как pd импортировать clickhouse_driver импортировать os

CLICKHOUSE_CLIENT = get_clickhouse_client()

default_args = { "start_date": datetime(2024, 1, 1) }

с DAG( dag_id="datamarts.daily_revenue_per_country", default_args=default_args, schedule_interval="@daily", catchup=False ) как dag:

transactions_sensor = S3KeySensor(
    task_id="transactions_sensor",
    bucket_key="data/transactions_{}.csv".format(datetime.now().strftime("%Y-%m-%d")),
    bucket_name="my-bucket",
    aws_conn_id="aws_default",
    timeout=600,
    poke_interval=30,
    mode="poke"
)

def extract_from_s3(**kwargs):
    df = pd.read_csv("s3://my-bucket/data/transactions_{}.csv".format(datetime.now().strftime("%Y-%m-%d")))
    kwargs["ti"].xcom_push(key="df", value=df.to_dict())

def load_to_raw_table(**kwargs):
    df = pd.DataFrame(kwargs["ti"].xcom_pull(task_ids="extract", key="df"))
    rows = [tuple(r) for r in df[["transaction_id", "user_id", "amount", "created_at"]].to_numpy()]
    CLICKHOUSE_CLIENT.execute( ... )

Ответ

В DAG видны проблемы с импортом S3KeySensor, вычислением даты через now(), глобальным клиентом БД и передачей всего DataFrame через XCom. Их нужно устранить до запуска.

Фрагмент обрезан и записан с переведёнными ключевыми словами, поэтому это не готовый Python-код. Для Airflow 2 импортируйте S3KeySensor из amazon provider, а ключ объекта задайте шаблоном интервала: data/transactions_{{ data_interval_start | ds }}.csv. Передайте этот же ключ задаче чтения вместо нового вызова now().

Клиент ClickHouse создавайте внутри задачи. Через XCom передавайте путь и метаданные, а не весь набор строк. Загрузка должна быть идемпотентной для повторного запуска; простой INSERT этого не гарантирует. В показанном наборе полей нет страны, поэтому для витрины revenue_per_country понадобится явный источник соответствия пользователя стране и правила валют/возвратов.

Источник: Документация Apache Kafka.