I have used airflow telegram operator for notifying me on some metric after some task. Today I check the operator task fail with this exception: TypeError: AsyncConnectionPool.__init__() got an unexpected keyword argument 'socket_options' Well I have researched some solution to…
I'm using Debian/Ubuntu, and I am trying to install an Airflow provider in my Python virtual environment: $ pip install apache-airflow-providers-mysql error: subprocess-exited-with-error × Getting requirements to build wheel did not run successfully. │ exit code: 1 ╰─> [24 lines…
I have a table tbl in Postgres with 50 million rows. The tbl has an index on column_1 and there are a lot of queries to this table like select * from tbl where column_1 = 'value' Each query returns…
Today is 8th of September 2023. I run Airflow in Containers. Worker, Scheduler, etc. are on. I have the following DAG: from airflow import DAG from datetime import datetime from airflow.providers.postgres.operators.postgres import PostgresOperator with DAG('user_processing', start_date=datetime(2023,10,12), schedule_interval='@daily', catchup=False) as dag:…
I'm trying to import dags from a blob storage to a managed airflow from Azure Data Factory. The ADF instance can access the blob storage through the System Managed Identity (I can browse the storage through ADF), but when i…
There is a valid TelegramOperator function: send_telegram_message = TelegramOperator( task_id="send_telegram_message", token="<token>", chat_id="<chat ID>", text="Some notification" ) It works that way: t1 >> send_telegram_message But I want it to notify me about failed tasks. I've tried to put it into default…
I'm on MacOS Mojave, Python 3.8.3, and pip 23.2. I made a py_env, activated it, and ran pip install apache-airflow==2.6.3' --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-2.6.3/constraints-3.8.txt". At first I had an error saying it couldn't find file "pybind11". I installed pybind11 by running python…
I am running an Airflow job to load data into a table. The task is : query a database -> get results in pandas data frame -> pass the result set to a worker processes -> each worker process process…
I am trying to create a emr-serverless application through the EmrServerlessCreateApplicationOperator but I keep facing the error botocore.exceptions.NoRegionError: You must specify a region. I am passing the region like below: create_app = EmrServerlessCreateApplicationOperator( task_id="create_spark_app", job_type="SPARK", release_label="emr-6.6.0", config={"aws_access_key_id":args["aws_access_key_id"], "aws_secret_access_key": args["aws_secret_access_key"], "aws_session_token":…
$ airflow webserver ____________ _____________ ____ |__( )_________ __/__ /________ __ ____ /| |_ /__ ___/_ /_ __ /_ __ _ | /| / / ___ ___ | / _ / _ __/ _ / / /_/ /_ |/ |/…