I am using Airflow on local environment to add a step in EMR after creation of cluster using below code block. def add_step(cluster_id,jar_file,step_a...

Tags for DAG are created automatically. But is there a way to create meaningful description for the tag? So user can not only use tags for searching b...

We have deployed airflow using the official helm chart and we are using KubernetesExecutor with git-sync. I already managed to launch a worker pod wit...

I have around 50TB of data in my database which needs to be extracted from this source database and loaded into a downstream system. This data source ...

Requirement: To be able to install the rpy2 library, as the code to be orchestrated with airflow uses it extensively Current Dockerfile FROM ubuntu:la...

I'm trying to execute python scripts and some shell commands in airflow dag based on a dictionary. The code where I'm trying to achieve this is below....

I deployed airflow on kubernetes using the official helm chart. I'm using KubernetesExecutor and git-sync. I am using a seperate docker image for my w...

My local workspace has csv file "online_retail.csv" that I try to upload. Goal: upload file from local to s3. I have been running this dag below as te...

I want to store connection in local files in dev context. I use docker compose to run airflow. I try this guide and have a local file conn.json : { ...

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 da...

I am working on a simple Airflow-Docker project where I want to implement a ML model, use Flask to expose the model on a port so that any user can use...

The following template for bucket_key works and it pulls the output value from task aTestPyOperator: sensor = S3KeySensor( task_id='check_s3_f...

I installed Airflow, both through Apache and Astronomer and wrote a really simple DAG with two tasks, each of which are BashOperators that call a Pyth...

I am trying to access XCOM value while learning Airflow, but every time, I get None returned. Below is the DAG code. I have also set the dependency, s...

I have an Airflow TaskFlow task that returns output: @task def extract(**context): return { "json": "values" } How do I pass that to the Postgres ...

I have a task in airflow that uses a DockerSwarmOperator to run a short-lived container. Once the container finished execution with no errors (DESIRED...

Python3, Airflow 2.2.2 I have several Airflow DAGs I use for my data pipeline. The main DAG ("rollups") consists of tons of SQL queries that get execu...

I have created a DAG file to schedule python and Apache Ni-Fi tasks. In order to control the Ni-Fi pipeline, I have followed the suggestions contained...

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://ra...

i have more dags (say 150dags) in my airflow account. My airflow version = 2.4.0 im using url and query parameter http://localhost:8081/api/v1/dags?li...

How do the parameters retries, retry_delay and retry_exponential_backoff interact? When there is a retry, when is it scheduled? Exponential backoff so...

I want to deploy an application to Airflow that accepts a config file as a parameter, pulls the git repository specified by said config, then transfor...

my Apache airflow version is 2.5.0. I want to access and use the values of config json set on triggering the DAG. I have tried the following given sol...

I have created a DAG to upload a local file into a personal S3 Bucket. However, when accessing http://localhost:9099/home I get the following error: F...

I'm running a distributed Airflow 2.4.0 setup using the official Docker image. All the containers use the same .env file and same version of Airflow i...