Questions tagged [airflow]
Apache Airflow is a workflow management platform to programmatically author, schedule, and monitor workflows as directed acyclic graphs (DAGs) of tasks.
8,156
questions
0
votes
0
answers
6
views
Unable to get epoch in milliseconds in Airflow
I have the below airflow code:
time_for_run_id = time.time()
run_id=int(time_for_run_id * 1000)
SQL_1 = "INSERT INTO dds_control.dds_control_table (job_run_id,job_start,job_status) VALUES('{}',...
0
votes
0
answers
4
views
Make Airflow custom operator compliant with Taskflow API
I just discovered the Taskflow API by watching the Airflow summit 2022.
It works fine, but I still heavily rely on "classic" operators, some of them I have written myself.
Is there ...
0
votes
0
answers
6
views
Not Able to Access Google Cloud Composer Stable API Through Compute Engine
I have a use-case i.e. to trigger cloud composer DAG from compute engine that does not have explicit external IP and all the traffic to the public internet is routed through Cloud NAT and in Cloud NAT ...
0
votes
1
answer
14
views
Access airflow dag run conf in UI
I have airflow 1.10.12 installed on a server and I’ve launched a dag with a trigger dag configuration which is json. I know how to access the config in my dag’s code, but I want to get it in Airflow ...
-1
votes
0
answers
10
views
How to use airflow to automate several scripts
I want to run about 3 python scripts and a jupyter notebook, but I want to automate them, where one runs and when it finishes, the next one runs, etc
I am struggling with how to do it in airflow. I ...
0
votes
1
answer
31
views
Operators not referencing the same global variable
I am defining a global variable timestamp with multiple operators referencing this variable. It seems that this variable is redefined when each operator is run? Below is a minimal reproducible example....
1
vote
0
answers
19
views
Where can I see the logs for failed airflow scheduler deployments?
I'm deploying Airflow with Helm (I'm using Rancher to do so)
I've set up git-sync to pull the DAGs from a private repo. Everything is working fine.
Now I want to switch to a different repo. I updated ...
0
votes
0
answers
15
views
Python/Airflow AttributeError: 'AwsLambdaHook' object has no attribute 'update_relative'
I have tried to break this down in several ways since this code snippet comes within an elif loops but it all seems to come down the the AwsLambdaHook giving me trouble. That is, even hardcoding ...
0
votes
0
answers
8
views
Airflow sync 2 buckets in s3 [closed]
Can I do this in airflow's python script to sync 2 buckets in aws s3?
task = BashOperator(
task_id='task',
bash_command='aws s3 sync s3://bucket1 s3://bucket2')
0
votes
0
answers
18
views
Launch DAG based on when table is updated
I have a 10-15 step DAG process that I need to launch after one bigquery table is updated for the current day. The table can be checked with a simple select SQL statement:
select * from sample_table ...
0
votes
0
answers
12
views
How to use virtual environment in airflow DAGS?
I built a project with python scripts and now I'm using Airflow with Docker to orchestrate it. I built the project in a virtual env and now i don't know how to tie the pieces together.
I used https://...
0
votes
0
answers
16
views
Schedule DAG to run daily for last 30 days
I have a DAG that runs daily at 11:00 AM. I want to pull the data for the last 30 days because it involves some running metrics who's value changes over time in the last 30 days. I am using Airflow 1....
0
votes
0
answers
10
views
airflow jobs stucked in running status
There is a strange phenomenon when using air flow. When I run the airflow scheduler, the unidentified jobs suddenly become running, and when I exit the scheduler, it turns into success.
I tried ...
0
votes
1
answer
18
views
Create Multiple Subdags in One DAG
I have in main_DAG, multiple tasks (which are subdags): subdag1, subdag2, subdag3. Whenever I try to create a flow like this:
main_DAG>>subdag1>>subdag2>>subdag3
I get this error:
...
0
votes
0
answers
8
views
Service airflow-cli does not work with new docker compose format
I use the Docker deployment of Airflow from their official docs page. When docker-compose changed to docker compose, it broke the airflow cli commands. It hangs on airflow-cli-1 starting when I try to ...
0
votes
1
answer
15
views
minio ModuleNotFoundError when running Airflow Docker image
I have downloaded the latest Docker image for the Airflow and am able to spin up the instance succesfully. On my local system I have installed minio server using homebrew on my Mac.
I have created a ...
0
votes
0
answers
16
views
Airflow externalTaskSensors to sense multiple dags with different schedule
I am trying to create dependency between multiple dags.
Lets say Dag_A, Dab_B and and running every day at 14:15 and 14:30 respectively.
now i want to run Dag_C which runs at 14:30 having 2 sensors ( ...
0
votes
1
answer
24
views
Docker-compose postgres cant init my tables in newly created database
I am trying to automate the INIT process in Docker so whenever I docker-compose up in an empty data directory it will create two DBS 1st one is Airflow and the second one is my test DB so for instance ...
0
votes
0
answers
19
views
Scheduling multiple spark streaming jobs and its dependency management
I have a requirement where I need to support adhoc runnable task which are spark streaming jobs.
This is job scheduling mechanism should support orchestrating , monitoring , rescheduling and ...
0
votes
0
answers
8
views
Pickle Error in Pyspark While Airflow Migration
I'm migrating from old Airflow to New Airflow Environment.
While migrating to new airflow env, I encountered pickle error.
I understood the object need to be pickled because of RDD operation closure.
...
0
votes
1
answer
33
views
Pandas to_csv creating file while running as script but not creating file while running as airflow dag
I have a below script which uses pandas to_csv to create a csv
dict_to_csv.py
import pandas as pd
import pathlib
import os
def create_test_csv():
sample_dict = {
'test1': (1, 2001),
...
0
votes
0
answers
22
views
Need help setting up a github repo
I am an EE student who was able to secure a summer job with a CS professor. He will start working on a development project and asked me to setup a GitHub repo for the same. I have coding experience ...
0
votes
1
answer
21
views
Airflow init in Docker creates empty directories
One of the projects I am working on is using airflow. So, I used airflow's documentation to install airflow with docker compose.
I wanted the dags plugin and logs directories to not be in the same ...
0
votes
0
answers
34
views
AIRFLOW trigger DAG (dynamically) multiple times
I have a DAG where data elements are generated and stored in a REDIS database.
When the storage of all elements are done, I want to trigger a further processing DAG.
Right now, I take the top entry ...
-1
votes
0
answers
19
views
airflow-name '_mysql' is not defined MacOS
I'am using airflow to connect to my docker mysql container, but exception of "name '_mysql' is not defined" when testing the connection. I can get connected by using the Dbeaver.
So what ...
0
votes
0
answers
26
views
DBT Airflow DAG JSON overrides using Kubernetes operator
Created a DAG to access overrides. Below are the commands that are passed in the overrides.
JSON on Airflow UI : {"dbt_command": ["products", "--full-refresh"]}
The ...
0
votes
0
answers
14
views
Airflow:Logs not showing in the UI while tasks running
Airflow Version: 2.2.4
Airflow running in EKS
Issue: Logs not showing in the UI while tasks running
The issue with the logs is that airflow is only writing the logs to the log file rather than ...
0
votes
1
answer
15
views
I want to pass arguments from dag to trigger another dag
The code below is a situation in which var1 and var2 are passed using the conf parameter when triggering another dag from the first dag.
trigger = TriggerDagRunOperator(
trigger_dag_id='dag2',
...
0
votes
1
answer
13
views
Problem with BranchPythonOperator to TaskGroup
I got the following DAG
import logging
from airflow import DAG
from airflow.operators.dummy import DummyOperator
from airflow.operators.python import BranchPythonOperator
from airflow.utils.dates ...
1
vote
1
answer
23
views
Pass variable to jinja template
I am quite new with Python and jinja and I would need some help about the following problem.
In order to deal with some variable in airflow, I need to define a template which involves a variable ...
0
votes
0
answers
19
views
Cloud Build run python script using custom container fails
I'm using a customer container apache/airflow from docker-hub to run a simple python script test_file.py and it executes without problem.
from airflow.models import BaseOperator
print('import works. ...
2
votes
0
answers
18
views
How can I download a file from an S3 bucket after I SSH to a remote machine/edge node to a particular path in the same remote machine?
I am relatively new to Airflow and am trying to solve a particular problem.
I wanted to know how I could download my file residing in an S3 bucket after SSH to a remote machine/client through the ...
0
votes
1
answer
27
views
Airflow using ExternalTaskSensor Operator caused MySQL innodb deadlock
I use ExternalTaskSensor Operator in Airflow to manage dependencies between DAGs, My ExternalTaskSensor Operator code looks like this:
dag = DAG(
dag_id='sushi.batch.load.application.detail.1d',
...
0
votes
0
answers
58
views
Running docker container as amd64 machine using a M1 Mac
I have a Docker file which creates an image and then I run it using docker compose together with a container built using a Postgres image. (To set up a local environment of Airflow - we use the mwaa ...
0
votes
0
answers
22
views
Error when running selenium on Apache Airflow- Issue related to ChromeDriver run-Giving Unkown error
I'm getting this error when running selenium on airflow. The airflow uses AWS to store the files, I used the following lines of code:
OP = webdriver.ChromeOptions()
# op = FirefoxOptions()
# op....
0
votes
0
answers
42
views
Cloud Composer, "Lost connection to MySQL server at 'reading initial communication packet', system error: 0"
I'm running the Cloud Composer environment with Composer version 1.18.7 and Airflow version 1.10.15. Based on Google Cloud Platforms documentation the error message "(_mysql_exceptions....
0
votes
1
answer
38
views
How can I get the results of a query in bigquery into a list?
I'm running a DAG that runs multiple stored procedures in Bigquery in each DAG run. Currently, my code is the following
sp_names = [
'sp_airflow_test_1',
'sp_airflow_test_2'
]
# Define DAG
...
0
votes
0
answers
16
views
Rest API requests failing in Airflow ('Connection aborted.', ConnectionResetError(104, 'Connection reset by peer'))
I'm requesting data through a rest API using the requests library. Everything works locally, but when I deploy to my GCP composer instance I experience errors.
def get_ids():
s = requests.Session()...
2
votes
1
answer
23
views
Where should I put the .pem file to connect to a EC2 instance via SSH from a containerized Airflow service?
I am having trouble connecting to a EC2 instance from a dockerized puckel/Airflow service. I don't know where to save the pem file for the SSH connection to grab. How can I make the pem file ...
0
votes
2
answers
48
views
Start a Docker container from Apache AirFlow without creating one
I have a running Apache AirFlow started with docker-compose. Also I have a docker-compose project which I'd like to manage with AirFlow.
example_workers/docker-compose.yml:
version: '3'
services:
...
0
votes
0
answers
28
views
How do I solve "No matching distribution found for apache-airflow==...+composer" when installing PyPI packages for my Cloud Composer instance locally?
I'm trying to set up a local development environment in VS Code where I'd get code completion for the packages Cloud Composer/Apache Airflow uses. I've been successful so far using a virtual ...
0
votes
0
answers
20
views
Setting up Airflow with Docker
I am trying to setup Airflow via Docker on AWS EC2 Linux 2. I am referencing this blog: https://towardsdatascience.com/setting-up-apache-airflow-2-with-docker-e5b974b37728
.
I have changed the ...
1
vote
1
answer
31
views
Airflow Branch Operator and Task Group Invalid Task IDs
I have a simple dag that uses a branch operator to check if y is False. If it is, the dag is supposed to move on to the say_goodbye task group. If True, it skips and goes to finish_dag_step. Here's ...
-1
votes
0
answers
25
views
Airflow and Pagerduty alerts [closed]
Let say I have to implement accurate alerts from pager duty on Airflow tasks that are based on SLA. How to make an architecture and process diagram for this implementation. I am also using GKE in ...
0
votes
1
answer
40
views
Airflow web: Pass program arguments to DAG as an array or list
I am pretty new to Airflow.
I have an airflow environment set up and all my services are being triggered fine by airflow. But, I intend to pass program arguments to one of the services defined in my ...
0
votes
1
answer
32
views
Apache Airflow 2.3.0: Dags are not appearing in the browser
Can any one please help me to resolve this error, that is appearing in the Airflow 2.3.0 console.
Do not use SQLite as metadata DB in production – it should only be used for dev/testing. We recommend ...
0
votes
0
answers
22
views
How to send an email in AWS MWAA (Apache Airflow) using EmailOperator
I am working with AWS MWAA (Apache Airflow). I want to send an email in MWAA upon completion of my pipeline.I have set the following configuration
Now when I run my dag using an Email Operator, it ...
0
votes
0
answers
14
views
How to use different config files for different environments in airflow?
I'm using SparkKubernetesOperator which has a template_field called application_file. Normally on giving this field a file's name, airflow reads that file and templates the jinja variable in it (just ...
0
votes
0
answers
20
views
Error while running the command airflow webserver
I am trying to run airflow apache in virtual environment in Ubuntu.
I am using the below version
Python 3.10.4
PIP 22.1
Airflow 2.3.0
I have installed airflow via WSL
I am currently facing the issue ...
3
votes
2
answers
37
views
How to stop Dataproc Job from airflow
From Airflow we can submit dataproc jobs via DataprocSubmitJobOperator.
We can stop dataproc jobs in development environment but not in Production environnment via GCP Console.
Is there any way, we ...