Questions tagged [airflow]

Apache Airflow is a workflow management platform to programmatically author, schedule, and monitor workflows as directed acyclic graphs (DAGs) of tasks.

Filter by
Sorted by
Tagged with
0 votes
0 answers
6 views

Airflow 2.6.2 /tmp folder keeps getting very large

Since I've updated to Airflow 2.6.2 the /tmp folder inside the Docker-Compose keeps getting very large. I have to delete it manually to keep Airflow running. What could be the root cause? How can I ...
Charix69's user avatar
0 votes
0 answers
9 views

Pass Arguments from Airflow-default_args to Snowflake Query in KubernetesPodOperator

We have a DAG in Airflow that extracts data from a database and writes it to a file. The DAG uses a KubernetesPodOperator task which calls another repository that has all the queries in .sql files. ...
Ctrl Alt Elite's user avatar
0 votes
0 answers
22 views

How can I know what operators use local resources on a cloud environment?

I'm fairly new to apache airflow, I've some dags running on AWS (Using MWAA) and created a python code to extract data from athena to S3 and use it on redshift. Now, we are migrating to an EC2 ...
Fábio Elias Reis Ritter's user avatar
0 votes
0 answers
11 views

Clearing xcom data with dag_id and execution_date

I am trying to clear xcom values based on the execution_date older than 30 days, but the dag is throwing error as execution date is not defined . Please help to understand the issue here (airflow 2.4....
dewdrops's user avatar
1 vote
1 answer
18 views

Aiflow Branch issue

There're 4 task to run, and a condition filier I expect that happen: if t0 return True, skip t1, run t2 then t3 if t0 return False, run t1, but when t1 done, run t2 then t3 logic like: enter image ...
Dan Ling's user avatar
-1 votes
0 answers
20 views

Postgresql function does not work when called from an airflow DAG

I have a function that deletes a partition of a given table. It works fine when called as is(for example from IDE), but it refuses to work when called from an airflow dag. Well kinda refuses, it ...
YKY's user avatar
  • 2,531
0 votes
0 answers
4 views

Airflow ExternalTaskSensor not Launch

I have 2 dags, when second dag should be launched when all tasks of dag 1 finish. I have this code: download_exchange_rate_dag_name = 'daily-currency-exchange-rate-v2' start_date = datetime.datetime(...
Miguel's user avatar
  • 171
1 vote
0 answers
17 views

How to gitignore parent folder while keeping subfolders to help python interpreter find created modules?

Assuming I have the following git repo for an Airflow app dags/ ├── dag1.py └── dag2.py custom_packages/ ├── package_1.py └── package_1.py .gitlab-ci.yml The problem is that in the airflow server the ...
Ranieri Bubans's user avatar
0 votes
0 answers
21 views

LocalExecutor or CeleryExecutor for Airflow on a single machine?

My question may sound dumb, but we have a ETL process running in a remote machine (EC2) and as we only use it internally in the company and there are only 5 of us, we set up Airflow with Docker ...
WLD's user avatar
  • 17
0 votes
0 answers
17 views

Unable to install google-re2 on macos 14.1.1

I want to install Apache Airflow and in the installation process, it needs google-re2. The error is: Building wheel for google-re2 (setup.py) ... error error: subprocess-exited-with-error × ...
Mahdi's user avatar
  • 947
0 votes
0 answers
11 views

Airflow AttributeError: 'Context' object has no attribute 'query'

I am writing an airflow dynamic dag python file . In the "end" task , i had passed an "on_success_callback" to clear the Xcom if data is older than 30 days . But it is throwing ...
dewdrops's user avatar
0 votes
0 answers
10 views

Airflow on Kubernetes: why schedulers stop queueing dynamically mapped tasks after some time?

I have an Airflow DAG with a dynamically mapped task. We run quite heavy loads there (>4k mapped tasks). For this DAG the max_active_tasks is set to 32. It may take couple of days to run completely ...
illyge's user avatar
  • 183
0 votes
1 answer
20 views

Airflow skipped dag run mail notification

Is there any option in AWS airflow to get alert mail for the skipped dag runs. In my case, for a dag there are four task and every time run is getting skipped at third task.i have added ...
Gayathri's user avatar
0 votes
0 answers
11 views

Airflow encode path to home dir

I'm using airflow deployed in docker. When I try to use the AIRFLOW_HOME environment variable in a BashOperator I get /home/*** instead of /home/airflow. Even if I use a path without an environment ...
Иван Викторов's user avatar
0 votes
1 answer
45 views

Pass runtime parameter to two dependent tasks in azure

I am triggering a job in workflow from airflow and passing two runtime parameters in the job . My first task is a wheel file and the second dependent task. below is the JSON API structure in ...
Parijat Chakraborty's user avatar
0 votes
0 answers
29 views

Airflow SQL Server stored procedure does not complete

We are moving some stored procedure from being executed in SQL Server jobs to Airflow. I use the following template to execute them: usp = MsSqlOperator( mssql_conn_id='mysqlconnection', ...
Thibaud's user avatar
0 votes
0 answers
37 views

Passing argument on SparkKubernetesOperator

I am using a spark with airflow, but not able to pass the arguments. I have tried multiple ways please suggest where right way to do this this. dag.py file: base_operator = SparkKubernetesOperator( ...
Shivam Gupta's user avatar
0 votes
1 answer
36 views

retain message in sqs even after read

I've a Airflow Dag, which runs for multiple tables processing, but for one table at a time. Dag instances run in parallel if need to run more than one table processing. My problem is, I need to send ...
morgan's user avatar
  • 29
0 votes
0 answers
28 views

Run DAG tasks in loop

I am new to airflow and have a scenario to pull off. I need to pull a list from xcom and then loop over a set of tasks. I found this answer and it works well. I still am not able to get the list from ...
Prashant singh's user avatar
0 votes
0 answers
12 views

How to add Access Control to Airflow Plugins that expose custom REST APIs

I'm using Airflow plugins to expose a custom REST API. This is done via Flask Blueprints and adding route to handlers. How can I add access controls similar to Airflow REST APIs for these custom APIs ...
Tijo John's user avatar
0 votes
0 answers
22 views

How to run shell script from aws ECS operator

I have runcustomscript.sh inside my container (abs path: /opt/scripts/runcustomscript.sh) and I want to execute it through ECS operator: EcsRunTaskOperator( ... overrides={ "...
safa's user avatar
  • 21
-2 votes
0 answers
28 views

Airflow vs Snowflake Tasks [closed]

One day I found this post: https://www.reddit.com/r/snowflake/comments/17193nd/what_tradeoffs_do_you_see_between/ and it occurred to me a question does snowflake is better than airflow? What airflow ...
GGG205's user avatar
  • 835
0 votes
0 answers
21 views

Airflow DAG: ETL_toll_data.py does not exist in 'dag' table

I tried to unpause an airflow dag with the python file "ETL_toll_data.py" that I created in one of my projects, but I got the error "DAG: ETL_toll_data does not exist in 'dag' table&...
Austine's user avatar
0 votes
1 answer
52 views

Amazon MWAA Environment - Maximum worker count Must be between 1 and 0

Hi I am new and trying to learn Airflow on AWS, I am creating the Environment and got the error that Maximum worker count must be between 1 and 0. Could someone please help me to fix it. Thank you AWS ...
Truong Giang Nguyen's user avatar
0 votes
0 answers
11 views

Airflow Scheduling by weekly/monthly how to pick the start date when I want the end date as processing date

My requirements is to schedule the Airflow Dag weekly/monthly. I have gone through the concepts of airflow like, it will execute the dag once the interval is completed, date_interval_end=...
timmanna's user avatar
1 vote
0 answers
33 views

Minikube on Docker shows high CPU utilization every 10 seconds

For development I am using the following setup: 8 core 16 thread CPU w/ 64 GiB RAM Arch Linux Docker Minikube on Docker (4 CPUs 16 GiB memory) This single node kubernetes cluster houses 1 RabbitMQ ...
reallynotthateducated's user avatar
0 votes
0 answers
37 views

Airflow branch operator doesnt recognize my task groups

i have the following dag, it creates dynamic tasks within task groups, based on mng tables in snowflake. the thing is, i cant seem to get the branch operator works with these task groups. it is ...
Asaf Rabi's user avatar
0 votes
0 answers
11 views

how do I enable debug logs for triggers?

I'm debugging an issue now, I have managed to enable debug logs for operators by updating airflow.cfg with [core] logging_level = DEBUG But I don't see any trigger logs in the task view. How can I ...
Dagang Wei's user avatar
0 votes
0 answers
16 views

Firefox Binary OK for Manual Run but Fails When Scheduled

I have a DAG in which the firefox binary path is specified: options = FirefoxOptions() options.binary = "/usr/bin/firefox" It executes consistently for manual runs (airflow dags test <...
Lars Skaug's user avatar
  • 1,376
0 votes
0 answers
17 views

Facing issue in airflow dag skipped dag mail notification

I wanted to know if there is anything like on_skip_callback to send notification. Using html_email_generator for mail notification. In my case, for a particular run there are four task and every time ...
Gayathri's user avatar
0 votes
0 answers
11 views

Replacement index 1 out of range for positional args tuple on Astro sql load_file package

I have installed astr cli recently and tried to load a csv file to gcp bigquery. I get a error File "/usr/local/lib/python3.11/site-packages/astro/databases/base.py", line 788, in ...
user546298's user avatar
0 votes
0 answers
37 views

How to write task dependencies in airflow which has conditional tasks [closed]

I need to have a Dag file which has conditional tasks in it and I generated tasks in the Dag file based on the attached image but struck with Task Dependencies. Kindly generate task dependencies and ...
Sarvath S's user avatar
  • 205
0 votes
1 answer
23 views

are airflow tasks scheduled based on the container time or the system time?

I am running airflow in a docker container on my local machine if I open a terminal and date it returns Wed Nov 22 18:36:44 EST 2023 if I do docker exec and run date inside the container it returns ...
Amin Ba's user avatar
  • 1,801
0 votes
0 answers
20 views

How can I run the downstream task depending on the state of each task generated by dynamic task mapping in airflow

I have to create a DAG with tasks that are dynamically generated through dynamic task mapping and do the subsequent actions depending on the state of each task. For example, what I want is a DAG with ...
Jisoo Song's user avatar
0 votes
0 answers
32 views

Invoke a aws lambda function using MWAA DAG and schedule it for 15 minutes

Invoke a aws lambda function using MWAA DAG and schedule it for 15 minutes Tried and followed below solution but it didnt work Python/Airflow AttributeError: 'AwsLambdaHook' object has no ...
Ram Sai Nandan's user avatar
0 votes
0 answers
29 views

How do I get a table from postgresql to be saved as a csv file in python running in airflow?

Apologies in advanced if the structure of the question is messy, as it is my first post here. I have this code below that runs without error if done in a code cell (Jupyter notebook), but when I tried ...
neranaiya's user avatar
0 votes
0 answers
22 views

Setting default password to airflow docker

In order to setup an easy-to-deploy local airflow environment for testing purposes, I wanted to create a simple docker-compose file using the airflow docker image. The resulting docker-compose is dead ...
Jérôme Fink's user avatar
0 votes
0 answers
14 views

Airflow DAG Not Starting as Expected on Scheduled Date

I'm experiencing an issue with scheduling a new DAG in Airflow. I set the start date for the DAG to 2023-11-22 (I did this on 2023-11-21 and this was synced through Git to Airflow), but one day later, ...
Siraxis's user avatar
  • 29
0 votes
0 answers
27 views

Airflow: Task-generated Mapping Over a Task Group: Branching Behavior Not Functioning as Expected

default_args = { 'owner': 'airflow', 'depends_on_past': False, 'start_date': datetime(2023, 11, 1), 'email_on_failure': False, 'email_on_retry': False, 'retries': 1, '...
Ethan Tan's user avatar
0 votes
1 answer
30 views

Airflow and LambdaInvokeFunctionOperator and payload

I'm using Airflow 2.7.3. I have a start-process task that pushes to Airflow Xcoms two values: schema (string), and ids (list of integers). Then I try to create a LambdaInvokeFunctionOperator like this:...
Kari Marttila's user avatar
0 votes
0 answers
10 views

Airflow - copying Sqoop jars

Going through the sqoop documentation here and found below Sqoop will copy the jars in $SQOOP_HOME/lib folder to job cache every time when start a Sqoop job. When launched by Oozie this is ...
luckyluke's user avatar
  • 644
0 votes
0 answers
19 views

Airflow - Dynamic task mapping and Multi output Xcoms

In Airflow (2.7.1) I would like to use the output of a task with multiple_outputs in a dynamic task mapping call: @task(multiple_outputs=multiple_outputs) def get_variable_key(variable): return ...
Flo's user avatar
  • 389
0 votes
1 answer
22 views

BeamRunJavaPipelineOperator settings in Airflow using GCP Dataflow (doing a non templated job!)

I'm struggling to get my BeamRunJavaPipeline() working in Airflow to run a Dataflow job on GCP. I already have the jar file in Google Cloud Storage. I'm basically looking for pointers and a practical ...
Sander van den Oord's user avatar
0 votes
0 answers
56 views

AWS MWAA conflicting dependencies between compatible versions requirements.txt

I am setting up a AWS Managed Workflows for Apache Airflow (MWAA) environment using airflow version 2.6.3, and I am using a requirements.txt file to install my desired python packages as suggested by ...
Hallow's user avatar
  • 1
0 votes
1 answer
63 views

Can I Create/List Pods in Azure AKS without Installing Kubelogin, Similar to EKSHook in Amazon, Using Airflow?

I'm currently working on a project where I need to create and list pods in Azure AKS using Airflow. I know that there is a way in Amazon using EKSHook without the need for an extra installation like '...
Jun Kim's user avatar
1 vote
1 answer
34 views

How set description to DAG tags in apache airflow?

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 but see description of tags that are used.
Cherry's user avatar
  • 31.7k
1 vote
0 answers
35 views

How can we connect to Microsoft Dynamics 365 CRM by using Airflow to retrieve data?

I am new to airflow and I am unable find any information regarding this question: how can we connect to Microsoft Dynamics 365 CRM using Airflow to retrieve data? I try to find out any link or Youtube ...
Venkata Lakshmi Narsingarao Ma's user avatar
-1 votes
1 answer
326 views

Airflow: from connexion.decorators.validation import RequestBodyValidator ModuleNotFoundError: No module named 'connexion.decorators.validation'

After installing Airflow in Aws or Azure Vm whenever I'm entering the command "Airflow standalone" it ends up with an error below. I have tried it in different instances of EC2 and Azure Vm ...
Nadir Hussain's user avatar
0 votes
1 answer
18 views

Airflow DAG: using returned values from previous task to create following tasks dynamically

I would like to calculate dates before I created next task, Ideally one task per date. I tried TaskInstance.xcom_pull() to access to returned value of first task. But when I tried to used that in a ...
user3145047's user avatar
0 votes
1 answer
39 views

I can't run any script of my project without a flag and I supect it is interfering on airflow working on my python env

I'm facing an issue that, until now, was manageable. After installing a local package in my conda environment, I needed to add a flag to run any script in my project, such as a simple data extraction ...
Lucas Vital's user avatar

1
2 3 4 5
206