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,132
questions
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?
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
# ...
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
...
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.
...
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',
...
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....
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. ...
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 ...
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. ...
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. ...
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)
...
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
...
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-...
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", "...
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 | ...
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 ...
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 ...
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 ...
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 ...
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 ...
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,...
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'...
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=...
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 ...
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 ...
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 ...
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:...
-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/...
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=${...
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 = {
"...
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 ...
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, ...
-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 ...
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 ...
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 = {
...
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 ...
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 ...
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 ...
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',
...
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 ...
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 ...
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 ...
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 ...
-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 ....
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 ...
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 ...
-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 ...
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 ...
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 ...
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 ...