def extract_from_s3(**kwargs):

Условие

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( "INSERT INTO raw.transactions (transaction_id, user_id, amount, created_at) VALUES", rows )

def build_aggregate_view(): query = """ INSERT INTO datamarts.daily_revenue_per_country SELECT toDate(r.created_at) as event_date, u.country, sum(r.amount) as total_revenue FROM raw.transactions r LEFT JOIN core.userMetadata u ON r.user_id = u.UserId WHERE toDate(r.created_at) = '{}' GROUP BY event_date, u.country """.format(datetime.now().strftime("%Y-%m-%d")) CLICKHOUSE_CLIENT.execute(query)

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" )

extract = PythonOperator( task_id="extract", python_callable=extract_from_s3, provide_context=True )

load = PythonOperator( task_id="load", python_callable=load_to_raw_table, provide_context=True )

aggregate = PythonOperator( task_id="aggregate", python_callable=build_aggregate_view )

transactions_sensor >> extract >> load >> aggregate

Ответ

Пайплайн не воспроизводим из-за now() и может удваивать raw и агрегаты при retry. Кроме того, дубли userMetadata способны размножить транзакции при JOIN и завысить выручку.

Используйте единый data interval для sensor, чтения файла и агрегации. Клиент БД создавайте внутри задачи; передавайте через XCom ссылку на данные, а CSV читайте порциями. INSERT raw должен иметь стратегию дедупликации transaction_id или замены проверенной порции.

Перед JOIN получите одну нужную версию пользователя: текущую либо историческую на момент продажи — это разные метрики. Пересборку агрегата делайте идемпотентно, с заменой соответствующего периода, а не повторным суммирующим INSERT. Согласуйте timezone, Decimal для денег и валюту. Вместо форматирования SQL передавайте параметры; настройки provide_context и sensor нужно сверить с версией Airflow.

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