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,458
questions
0
votes
0
answers
3
views
Upgrade python version in AWS Apache Airflow (MWAA)
I am working with Apache Airflow in AWS (MWAA) and I have deployed a python package that was developed on python 3.8. It was running fine locally, however, on MWAA it starts giving syntax errors ...
0
votes
1
answer
11
views
Hot to enable live reload on docker-compose for Airflow?
This is probably a duplicate, but couldn't find it here - if yes, sorry for that.
I have a simple docker-compose Airflow app with docker-compose.yml:
services:
postgres:
...
init:
...
...
0
votes
0
answers
13
views
creating sequential workflows using and ETL tool (Fivetran/Hevo), dbt and a reverse ETL tool (Hightouch)
I'm working for a startup and am setting up our analytics tech stack from scratch. As a result of limited resource we're focussing on using 3rd party tools rather than building custom pipelines.
Our ...
1
vote
0
answers
32
views
Airflow: mistake "decoding with 'utf-16le'" when using several connections
Have a problem with my ETL process.
I've got ETL process, written in python and it works great, but operations
starts one after another, so the whole process lasts much time.
I'm slightly new in ...
0
votes
0
answers
10
views
Is there any way where Airflow can pick my DAG files from s3 directly?
Airflow can only pick DAGs from either "/dags" folder or in the particular Github repository.
Any other ways we can make Airflow pick DAG's from?
0
votes
1
answer
15
views
How to only run certain operator when dag conf value exist
def update_job_pod_name(dag: DAG, job_id: str, process_name: str) -> MySqlOperator:
"""
:param dag: Airflow DAG
:param job_id: Airflow Job ID
:param process_name: ...
0
votes
0
answers
16
views
Airflow run bash command on an existing docker container
I have Airflow running in a Docker container and I want to trigger a python script that resides in another container. I tried the regular bash operator but that seems to be only for local. Also ...
0
votes
0
answers
24
views
On Failure to check which file is failed in gcp
We know that using * can load n number of files available in the bucket directory.
gs:/bucket/file*.csv
My question during data load how do we find which file load is failed.How do we track
0
votes
1
answer
21
views
importing python dask package on aws MWAA airflow not working
I am trying to import dask package on AWS MWAA (airflow) using requirements.txt, by adding "dask==2022.6.0" line to the file.
Not only the import does not work, the other packages are not ...
1
vote
0
answers
33
views
Can we run pyspark on a remote server from apache airflow?
My organisation's apache airflow is installed in the server let's assume abc. I have a piece of pyspark code that used to run on a different cluster with master on a different server let's assume xyz. ...
0
votes
1
answer
19
views
When to create a custom operator in airflow?
So there is a use case where I want to check the size of file and if it more than the size of a configurable value (changes per file type as per business rule) , we would return true/false as a return ...
0
votes
1
answer
24
views
Scheduling SQL Queries inside Airflow DAG
I have a use case where we extract SQL Queries from a REST Endpoint and schedule them using an external scheduler, Airflow in our case. So I was thinking of solving this by doing the following steps.
...
0
votes
0
answers
21
views
Problems connecting Redshift to Airflow (MWAA)
I am learning airflow and as a practice exercise im trying to create a table at Redshift through an airflow dag at MWAA. I create the connection to Redshift at the UI (specifying host,port, etc) and ...
0
votes
1
answer
19
views
Airflow mandatory task execution Trigger Rule for BranchPythonOperator
I have to create Airflow DAG with these states:
Task1 which is a BranchPythonOperator which decides to execute either Task2 or not and then execute Task3 irrespective of Task2 is ignored or failed or ...
0
votes
1
answer
35
views
Airflow directories, read a file in a task
I've discovered Airflow recently and I want to do a couple of simple examples to know how it works.
So I've build a very simple DAG (example_airflow.py) like this
import airflow
from airflow import ...
0
votes
1
answer
35
views
Airflow Passing date Xcom variable to a sql file using RedshiftSQLOperator
I want to pass the date value (from a variable) as partition column in sql file. The variable is not getting evaluated. The variable is stored in Xcom. Is there any better way to do .
Rendered ...
0
votes
1
answer
21
views
MS SQL Hook and Operators not importing into Airflow
Trying to import the mssql hook and operator into my dag but I keep getting this error from Airflow.
I'm currently importing with the newest syntax:
from airflow.providers.microsoft.mssql.hooks.mssql ...
0
votes
0
answers
31
views
Cloud Composer 2 unable to send SMTP emails
I am struggling in sending a email from cloud-composer 2 environment(composer-2.0.11-airflow-2.2.3) via smtp.gmail.com.
Please find my configs below :
Note I tried using smtp_password_cmd and ...
0
votes
0
answers
33
views
BigQuery / Airflow - Cannot create partitioned table
I'm using the BigQueryCreateEmptyTableOperator within Airflow to create tables in my BigQuery project. It's been working great, for everything except partitioned tables.
When I try to pass the ...
0
votes
1
answer
29
views
how schedule a job in airflow for first 4 business day (not on weekends) of the month
job should run on first 4 business day(not on weekends) at 3 am
ex : 00 03 1-6 * 1-5 ==> jobs are running every day-of-week from Monday through Friday
1 - fri 03:00
4 - mon 03:00
5 - tue ...
1
vote
1
answer
13
views
ExchangeImpersonation SOAP header must be present for this type of OAuth token in Python
Recently Microsoft has withdrawn the basic authentication service for accessing the Exchange Service. Therefore I am working on switching the authentication from basic to OAuth based. However, during ...
0
votes
0
answers
34
views
Error while retrieving data from a MSSQL db with python [closed]
I have a function to get data from MSSQL but I keep getting an error that is not very clear.
Here's my code:
def retrieve_data(**context):
dag_timer = time()
conn_pre = JdbcHook(jdbc_conn_id=SRC_CONN)
...
0
votes
0
answers
15
views
Airflow docker compose stuck
I am trying to run Airflow on my local machine. I am using docker compose file from official Apache Airflow webstiste https://airflow.apache.org/docs/apache-airflow/2.0.1/docker-compose.yaml
# ...
0
votes
0
answers
28
views
Airflow Password Variable
I have airflow variable named as mysql_password and set its value which is visible as **** after saving. When I try to access it using Variable.get, it still returns the **** value instead of the ...
0
votes
0
answers
12
views
Error while inserting data with a DAG file
I got a DAG file that is supposed to insert some data in an Azure SQL db.
I've created others and used this line: cursor.fast_executemany = True
When I run my DAG without it, it works fine but takes 7 ...
0
votes
1
answer
61
views
How to properly add operators based on array result retreived from another operator in Apache Airflow?
I need to first read from BigQuery to get a list of SQL strings stored in a specific table then loop through the string array and run against BigQuery again to fix things. For running those SQLs I ...
0
votes
1
answer
32
views
Branching in Apache Airflow using TaskFlowAPI
I can't find the documentation for branching in Airflow's TaskFlowAPI. I tried doing it the "Pythonic" way, but when ran, the DAG does not see task_2_execute_if_true, regardless of truth ...
0
votes
1
answer
48
views
How to use awk command in DAG file of apache airflow
I am trying to use awk command in DAG file to convert fixed width file to csv file but the command is not working. But it is working fine in ubuntu terminal
'''
extract_data_from_fixed_width=...
0
votes
1
answer
31
views
Airflow how to use input variable as default arg unless input variable not found
In Airflow, is it possible to use an input variable as a default_arg, but also have that arg safely resolve to an alternative if that input variable isn't given?
I'm trying for something like this:
...
0
votes
1
answer
47
views
MWAA/Airflow: Importing task code into dag files from submodules
I'm using MWAA on AWS for a workflow orchestration project. I have a DAG folder structure like this (and gets uploaded into S3 like this, with the MWAA Dags folder set to src/dags):
src/
├─ dags/
│ ├...
0
votes
0
answers
21
views
Airflow how to connect the previous task to the right next dynamic branch with multiple tasks?
I am facing this situation:
I have generated two dynamic branches. Each branch has multiple chained tasks.
This is what I need Airflow create for me:
taskA1->taskB1 taskC1->taskD1
...
0
votes
0
answers
29
views
Apache Airflow on Composer 2.0 Task Failed with "Not yet started" issue
My dags consists of too many tasks around more than 600 and while running the dag at times it behaves differently i.e sometimes a task gets failed and status comes as 'Not yet started'.
Please help ...
0
votes
1
answer
19
views
Does Airflow clear Tasks remove data from the database also?
We have a use case where we have to populate fresh data in our DB. There is already old data present from the successful DAG run in our DB . Now we need to delete the old data and re run the task.
...
0
votes
0
answers
58
views
Insert into BigQuery using parameters from web server request json data
In the google function I'm triggering a dag:
endpoint = f"api/v1/dags/{dag_id}/dagRuns"
request_url = f"{web_server_url}/{endpoint}"
json_data = {"conf": {"...
1
vote
1
answer
29
views
Airflow SQLOperator/Sensor vs MsSqlOperator
For apache-airflow 3.2.0
I am looking to use an operator to connect to an MsSql database. I found both SQLOperator and MsSqlOperator.
Does both work for an MsSql database or is it only the ...
0
votes
0
answers
28
views
Airflow logging doesn't show up when log from inside operator
I'm trying to log from inside an function in a dag, but it hasn't been working. I tried using print("something") as well but that didn't print any out either in the airflow log. How should I ...
0
votes
2
answers
61
views
BigQueryInsertJobOperator call Stored Procedure error
I'm having an error when trying to call an SP in BigQuery with Airflow BigQueryInsertJobOperator. I have tried many different syntax and it doesn't seem to be working. I pulled the same SQL out of the ...
0
votes
2
answers
40
views
Authenticating to a REST endpoint inside Airflow DAG
I need to extract data from an external REST API inside an Airflow DAG. The API is protected so I need to first authenticate/login to the API as a user, and then extract data by passing an ...
0
votes
1
answer
36
views
Airflow webserver fails with NoneType error
I'm having trouble running Airflow on a local k8s (Minikube), using the official Helm chart.
This is my custom Dockerfile:
FROM apache/airflow:2.3.1-python3.10
WORKDIR ${AIRFLOW_HOME}
USER airflow
...
0
votes
0
answers
31
views
Airflow - Test a TaskGroup
Is there a way to test all tasks within a TaskGroup in Airflow without running each task separately?
I have multiple task groups within a dag. I don't want to test the whole dag. I also don't want to ...
0
votes
0
answers
44
views
Packages version conflict
Hello I'm creating a dag in airflow that needs google-ads package but I'm getting this issue:
import sys
import os
import pandas as pd
from google.protobuf.json_format import *
from google.ads....
-1
votes
1
answer
38
views
Bigquery job failed with error: Encountered " "FROM" "FROM ""
I'm calling a SQL query with a BigQuery API with Airflow. This query works perfectly fine in the BigQuery workspace but says I'm writing FROM FROM even though I'm not...
The logs say line 4, character ...
0
votes
0
answers
26
views
How to include gcp_conn_id argument while using gcloud commands (in bash operator)
I am trying to use composer to create dataproc cluster. While I use DataprocClusterCreateOperator or DataprocCreateClusterOperator in my dag code, I am able to pass this argument gcp_conn_id='' to ...
0
votes
1
answer
26
views
How to setup remote debug for Airflow running with official docker-compose?
I have tried to follow this article: https://medium.com/@andrewhharmon/apache-airflow-using-pycharm-and-docker-for-remote-debugging-b2d1edf83d9d.
The problematic parts are:
Which container should I ...
0
votes
1
answer
40
views
EDIT: Airflow GCSToS3Operator: keep_directory_structure=True is not being passed in the rendered template, prefix is being added in the dest_s3_key
I am running the task in composer with Composer version: 2.0.18 and Airflow version: 2.2.5
I am sending data to the AWS S3 from Google GCS. For which I am using GCSToS3Operator with the parameters as ...
0
votes
0
answers
18
views
Making a dataset/table and inserting data into BigQuery using Node.JS
tokenConnection.js
const {GoogleAuth} = require('google-auth-library');
/**
* Instead of specifying the type of client you'd like to use (JWT, OAuth2, etc)
* this library will automatically ...
0
votes
1
answer
21
views
Error with Apache Airflow while running airflow webserver as daemon
I am running Apache Airflow 2.3.0 on an Oracle Linux machine, when I run the commands on 2 different terminals (commands airflow scheduler and airflow webserver), everything works fine. But, when I ...
0
votes
1
answer
21
views
airflow successeful task rerun before moving to another task
I am new to airflow and i am trying to have a task that is running every minute run alone for a while before going forward with other tasks. So my dag has a schedule interval of every minute and i ...
0
votes
1
answer
22
views
What is the relationship between airflow.cfg and airflow_local_settings.py?
Both of these files seem to define configuration. What is the difference between the two?
airflow_local_settings.py
airflow.cfg
0
votes
1
answer
41
views
Skipping task on failure in Airflow
We are creating tasks to load data from GCS to big query based on dativize sequentially. Consider
task a(03rd June)>>task b(04th June)>>task c(05th June) .
If task a is failed ,we don't ...