Questions tagged [airflow]
Apache Airflow is a workflow management platform to programmatically author, schedule, and monitor workflows as directed acyclic graphs (DAGs) of tasks.
10,294
questions
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 ...
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. ...
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 ...
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....
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 ...
-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 ...
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(...
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 ...
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 ...
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
× ...
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 ...
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 ...
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 ...
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 ...
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 ...
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',
...
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(
...
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 ...
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 ...
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 ...
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={
"...
-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 ...
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&...
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 ...
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=...
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 ...
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 ...
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 ...
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 <...
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 ...
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 ...
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 ...
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 ...
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 ...
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 ...
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 ...
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 ...
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, ...
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,
'...
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:...
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
...
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 ...
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 ...
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 ...
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 '...
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.
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 ...
-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 ...
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 ...
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 ...