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
1 answer
25 views

Trigger Airflow DAG on specific times daily

How should I pass the schedule_interval parameter in order to trigger my DAG everyday in specific times, such as at 10:30, 13:00 and 14:00?
gabikripka's user avatar
0 votes
0 answers
20 views

How can I render jinja template f-string?

Here is my test code: test_jinja.py i = 10 aa = f"{{{{ execution_date.subtract(hours=8).add(hours={i}).format('YYYY-MM-DD_H') }}}}" main.py import jinja2 import pathlib import pendulum # ...
이진환's user avatar
0 votes
0 answers
10 views

jinja2 render string multiple lines - expected token 'end of print statement', got 'string'

a.py test7 = ( '{{ execution_date.in_tz("Asia/Seoul").subtract(hours=1)' '.format("YYYY-MM-DD HH:mm:ss") }}' ) main.py import jinja2 import pathlib import pendulum ...
이진환's user avatar
0 votes
0 answers
11 views

What is the use of DataprocGetBatchOperator?

From the Dataproc and Airflow documentation, I couldn’t get much information about what is the exact need of DataprocGetBatchOperator. Whey I tested it, it return, config values of the operator. ...
Dishu Patel's user avatar
2 votes
2 answers
27 views

How to do nested templating using Jinja in an Airflow DAG?

The following template for bucket_key works and it pulls the output value from task aTestPyOperator: sensor = S3KeySensor( task_id='check_s3_for_file_in_s3', bucket_name='my_bucket', ...
Raj Rao's user avatar
  • 8,902
1 vote
1 answer
22 views

Use result of SqlSensor query as input to subsequent tasks

I am new to airflow and am trying to solve the following problem. We have a mysql database, and we want to export records to a files as they become an available. I have the SQL query I need to execute....
user3299166's user avatar
0 votes
0 answers
7 views

Airflow Logs Not Showing in CloudWatch for ECS Deployment

Description: I'm encountering an issue with Airflow when deploying it on AWS ECS and using CloudWatch for remote logs. Specifically, for some of our DAGs, the logs are not showing up in CloudWatch. ...
sel-nef's user avatar
  • 59
0 votes
0 answers
20 views

Start Date set in DAG is evaluating to a day earlier

In my DAG I am setting the start_date as 2023-09-26 and then when writing a test case, it is coming up as 2023-09-25 and hence failing the test case. I understand that this has something to do with ...
Matthew's user avatar
  • 331
0 votes
0 answers
23 views

How to import and create unit test for tasks in apache airflow?

I am a complete beginner in airflow and just created a basic airflow. So, my question is, inside the DAG where we are assigning tasks using PythonOperator, can we create unit test cases for that task. ...
Aatr Sin's user avatar
0 votes
0 answers
24 views

In Docker Airflow ModuleNotFoundError, module works fine in IDE

error I get the following error in airflow: ImportError: Missing optional dependency 'openpyxl'. Use pip or conda to install openpyxl. The issue is, I do have the module installed in the proper path. ...
Richard Willems's user avatar
0 votes
0 answers
13 views

Docker Airflow and python virtual environment

I have apache airflow that is running in a docker environment I have DAGS that where developed for different virtual python environments (different package compositions, different python versions) ...
user1573820's user avatar
0 votes
0 answers
13 views

Integrate Pagerduty On Airflow

i've been trying to send a notification from my airflow DAGs to Pagerduty, But it seems there's a missing code that triggers my DAGs to send notification to Pagerduty. Here's my code: import logging ...
Yohanes Lim's user avatar
0 votes
0 answers
11 views

Issue encountered while trying to run a process in Apache Airflow using Docker

Even though it ran successfully locally. The process involves using the pdfkit module directly, with the following code: pdfkit.from_string(text_content, filename, options={"enable-local-file-...
Shardon Paras's user avatar
0 votes
1 answer
16 views

How to pass a nested dictionaries as env variables to Kubernetes Pod Operator

In dag script PARAM = { 'key1' : 'value1', 'key2' : 'value2' } t1 = KubernetesPodOperator( task_id=task_id, name=task_name, cmds=["pipenv", "run", "...
zZzZ's user avatar
  • 143
0 votes
0 answers
18 views

How to determine the practical character limit of a Livy request to submit Spark batch job sent from Airflow?

I am trying to figure out the upper limit (practical estimate is fine) of characters/bytes passed on as args to the LivyOperator in Airflow. The operator eventually uses args(: Sequence[str | int | ...
gCW's user avatar
  • 1
0 votes
0 answers
15 views

HDFS Conection type Airflow not availlable in the list

I'm installing airflow on Openshift using Helm Chart, I made some changes on the docker image to add hdfs connection, I installed this package apache-airflow-providers-apache-hdfs and I checked that ...
Mohamed Ait Taleb's user avatar
0 votes
0 answers
13 views

How to extract tasks status report in tabular format of a Dag in Airflow? [closed]

I have to prepare one excel report of tasks status of a Dag with start time ,end time and duration. But Airflow Gantt option only shows the pictorial format and its difficult to manully prepare that ...
user22605668's user avatar
0 votes
0 answers
15 views

Airflow Variables timestamp incorrect updates [closed]

I am trying to save the timestamp of the extract_database task in query_timestamp and use it to update the airflow variable last_execution_timestamp as you can see. When I check the airflow variables ...
Mek Wiset's user avatar
0 votes
0 answers
46 views

dynamic pipeline based on result from sql using Airflow: TypeError: 'XComArg' object is not iterable

I wanted to make dynamic pipeline based on the result from query. I am trying to get some data from database and based on the result, I wanted to call different tasks. Why results from fetchall is not ...
Nandan Thakur's user avatar
0 votes
0 answers
27 views

Connecting SQL database to Airflow in Google Cloud Composer

hope all are well. I have a DAG running in the Google Cloud Composer. But, I am having an issue connecting sql db for the DAG. One way I tried was to export my local SQLite db file to Google Cloud ...
knk's user avatar
  • 37
0 votes
0 answers
13 views

Dockerized Airflow couldn't run Spark jobs

I tried to run a Pyspark task through executing python /myfile.py via BashOperator on Airflow but received the following error: [2023-09-19, 23:28:26 UTC] {subprocess.py:86} INFO - Output: [2023-09-19,...
Jonathan Feng's user avatar
0 votes
0 answers
8 views

Sync AWS CodeCommit with gitsync on SparkApplication+k8s

I'm running a spark-operator on k8s and I need to synchronize my AWS CodeCommit repository directly so I can import my python modules and not have to build the images with them encapsulated in it. I'...
Michel Senra's user avatar
0 votes
1 answer
18 views

Is there a way to "bundle" together airflow operators for re-usability?

I have 3 custom operators AOperator, BOperator, COperator. They are commonly used like this - a_data = AOperator( task_id="a_data", project_id=PROJECT_ID, region=...
Brian Jones's user avatar
0 votes
0 answers
12 views

Airflow is not able to work with Google dependecies

I'm using Airflow that is deployed using Docker I installed the latest version of apache-airflow-providers-google also tried(10.7.0) but faced with the issue Airflow is not able to use Google hook for ...
Paul's user avatar
  • 31
0 votes
0 answers
26 views

Dynamic Date Parameter in Airflow DAG for Snowflake SQL Query

I am using Google Cloud Composer with Airflow to run a SQL query on Snowflake. I need to set a date parameter based on a condition and then use that parameter in my SQL file. Specifically, I'm trying ...
MC7836's user avatar
  • 84
0 votes
2 answers
89 views

Unable to run a simple Airflow DAG with BashOperator calling Python scripts

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 Python script. The first Python script, in turn, reads ...
CodingInCircles's user avatar
0 votes
0 answers
17 views

Wrong Color Encoding In Airflow Logs

I am using Airflow task.virtualenv but my logs on the UI shows some weird characters from the background color encoding like [0m. Here is part of the log. [2023-09-18, 14:05:42 EDT] {process_utils.py:...
acai's user avatar
  • 459
-3 votes
0 answers
15 views

I have recently upgraded the cloud composer version and since then every time when the DAG gets triggered at the scheduled time I get following error

Log file is not found: bucket path 1 The task might not have been executed or worker executing it might have finished abnormally (e.g. was evicted). Please, refer to https://cloud.google.com/composer/...
Shifa Jada's user avatar
1 vote
0 answers
38 views

Can't install pytorch 1.13.1 in airflow 2.2.5

I am trying to install pytorch 1.13.1 in airflow-2.2.5 docker image via a Dockerfile like this: FROM apache/airflow:2.2.5-python3.9 ARG AIRFLOW_HOME_ARG=/opt/airflow ENV AIRFLOW_HOME=${...
dmayilyan's user avatar
0 votes
0 answers
17 views

Give access to a set of DAGS to a "builtin readonly role"

I am trying to follow https://flask-appbuilder.readthedocs.io/en/latest/security.html#role-based to give a user role access to certain DAGs. While the example FAB_ROLES FAB_ROLES = { "...
N1ngu's user avatar
  • 2,923
0 votes
0 answers
19 views

No module name "Custom packages"

my_airflow_project/ ├── dags/ │ ├── my_dag.py │ ├── another_dag.py ├── plugins/ │ ├── operators/ │ │ ├── my_custom_operator.py │ ├── sensors/ │ │ ├── my_custom_sensor.py ├ Call the ...
31 NQ's user avatar
  • 1
0 votes
0 answers
20 views

smtp_default connection is not working in AWS Airflow service

I am using AWS Airflow 2.2.2 version in our project. I am using SMTP for sending emails for which I have ben using environment variables as follows - Due to following deprecation warning in the logs, ...
Pallavi's user avatar
  • 548
-4 votes
0 answers
39 views

numpy.core._exceptions._ArrayMemoryError: Unable to allocate 73.4 GiB for an array with shape (4806656, 2050) and data type float64

My problem: I am running a task on Apache airflow and got the above mentioned error. I have an Windows system and Airflow is installed on docker. What I tried: From what I searched from web, a ...
Chhandosee Bhattacharya's user avatar
1 vote
1 answer
34 views

How to dispatch numerous xcoms from a task to another task?

I am brand new to Airflow and I am trying to understand data dispatch between tasks. The documentation for xcom (and a good article) helped me to understand the basics of a 1:1 data flow. Consider now ...
WoJ's user avatar
  • 27.3k
0 votes
1 answer
50 views

Terraforming a managed instance of Airflow on Azure

I'm trying to create a managed instance of Airflow on Azure using Terraform. This is what I have: terraform { required_version = "~> 1.3" required_providers { azurerm = { ...
csukcl's user avatar
  • 121
0 votes
0 answers
42 views

How to handle SSL CERTIFICATE_VERIFY_FAILED in Python to download data properly

To sum up what I'm working on: I need to run a DAG on Airflow that'll download some files from a public dataset, an american agency. Even after some days reading and testing plenty of solutions ...
Fábio Antunes's user avatar
0 votes
0 answers
25 views

Airflow 2.4 : UpStream_Failed Issue

I have created a dag in airflow that generates dynamic tasks on runtime. The DAG gets the configuration from two different tables and generates the tasks accordingly. Below is the DAG: The issue is ...
Nabeel Khan Ghauri's user avatar
1 vote
1 answer
17 views

Airflow DAG for detecting file drop in GCP bucket not working

I am attempting to trigger a dummy process from dropping a file in a bucket. I am monitoring this bucket using the GoogleCloudStorageObjectSensor. I run the DAG manually and the ...
climateBen's user avatar
2 votes
0 answers
36 views

How do I use airflow's FileSensor glob matching to find a file with a file with a specific string in its name and a set file extension e.g .txt/.csv?

I need to find a file being moved into a folder that has a specific name and also the .txt extension. I have seen both of these examples used: file_check = FileSensor( task_id='check_file', ...
Owen's user avatar
  • 23
0 votes
1 answer
21 views

Cross dag data exchange using Xcom

can We use Xcom to exchange data between two different Dags? So that My second dag should get the value from latest job of first Dag. So that My second dag should get the value from latest job of ...
Majeed's user avatar
  • 21
0 votes
0 answers
30 views

Gunicorn worker getting exit in airflow web server with Received signal: 15. Closing gunicorn

Gunicorn worker getting exit in airflow web server with Received signal: 15. Closing gunicorn. while installing airflow on kubernetes. Scheduler and other DB migration and user sync pod are running ...
Indrajeet singh thakur's user avatar
0 votes
0 answers
16 views

Logs saved locally and on AWS S3, but not loading in Airflow UI. (Airflow version 2.5.1)

I'm experiencing an issue with my Apache Airflow setup where the logs are being successfully saved both locally and on AWS S3. However, when I try to view these logs in the Airflow UI, they do not ...
Sagar Bhatt's user avatar
0 votes
1 answer
10 views

How to dynamic task mapping multiple parameters as pair, not cross product using taskflow api?

I would like to have parameter unpack in pair, set (if there is more than 2), not cross product. @task def add(x: int, y: int): return x + y added_values = add.expand(x=[4, 8], y=[40, 80]) # I ...
Haha TTpro's user avatar
  • 5,147
-2 votes
0 answers
24 views

Airflow Error-failed to load all files----Ignoring dags .... also found in

When I enter Command airflow dags list Found Error: Failed to load all files. For details, run airflow dags list-import-errors and When enter the command above terminal shows below error related to ....
atakshat11's user avatar
0 votes
0 answers
33 views

~10 min web post from within airflow operators getting stuck/hanging

Airflow operators are getting stuck when they include a web post that takes ~10 minutes. The following operators failed: PythonOperator that includes code to send an http post request with the ...
Maile Cupo's user avatar
0 votes
1 answer
43 views

Airflow - Executor reports task instance finished (failed) although the task says its queued

I recently upgraded my MWAA env from 2.4 to 2.6 (latest). I notice there was a breaking change in between the versions which I have updated. Otherwise, nothing has changed in my DAGs. However, almost ...
AIViz's user avatar
  • 92
-1 votes
0 answers
11 views

Airflow - Notification on broken dag

Can we impletement the notification on airflow , if DAG is broke , due to missing python lib and other reason . can you pls help how we can impletement this ?? we alredy have notifction if any DAG ...
sunil's user avatar
  • 1
0 votes
0 answers
9 views

how to use airflow worker to interact other container?

As I am install airflow in docker, but it's not possible to execute a operator from worker, to another container, how to solve the issue as "Got permission denied while trying to connect to the ...
Nick Chan's user avatar
0 votes
1 answer
21 views

Trying to run ECS Task using airflow scheduler

I am trying to run ECS task with the help of airflow scheduler. Just using the example template with few changes. But my task is running indefinitely when i execute DAG manually from airflow webserver ...
android2600's user avatar
0 votes
1 answer
21 views

Airflow dag hanging on running state when interacting with AWS S3 bucket

Simple Airflow DAGs are working for me, however, when I try to interact with an S3 bucket, the DAG will just hang on a running state. I'm using airflow S3 hook. If I run the code it will work but as a ...
d789w's user avatar
  • 357

1
2 3 4 5
203