Is it possible to make Python VirtualEnv Operator aware of local modules ? #22175
-
|
Hello, I want to import a scraper script I've done with its own dependencies as an Apache Airflow DAG. But here's the thing, virtual env is a different environment where I can't access my local modules (this is not a bug, it's expected). Still, how can I do to solve my issue ? Should I publish every script on pip, so I can import it as a requirement ? Is there any official hack ? # amazing_scraper_dag.py
@dag(dag_id= 'amazing_scraper', schedule_interval='@daily', start_date=datetime(2021, 1, 1), catchup=False, tags=['scrapper'])
def taskflow():
@task.virtualenv(
system_site_packages=True,
requirements=[
'aiohttp',
'asyncio',
'bs4',
'pandas',
'lxml',
]
)
def extract():
"""
#### Extract task
ERROR Here : Module solvolabs not found :'(
"""
# This function has my task requirements as direct dependencies
from solvolabs.my_scraper import scrape
return scrape()
extract()
dag = taskflow()I've found a lot of posts everywhere on GitHub and many hacks, but I'd like something more ... solid if possible. |
Beta Was this translation helpful? Give feedback.
Replies: 3 comments 14 replies
-
|
Package your This is the only approach that is not a hack that makes sense I think. |
Beta Was this translation helpful? Give feedback.
-
|
Thanks, it make sense to me. I'm worried about how I can make a package only available for my company but it's another issue not related to Airflow. Have a nice day 🙂 |
Beta Was this translation helpful? Give feedback.
-
|
You can use any "pypi"-aware artifact systems - for example artifactory supports PyPI-compliant package repository. |
Beta Was this translation helpful? Give feedback.
-
|
Hi ! Sorry to come back here : I've finally packaged my script and published it on PyPi : https://pypi.org/project/globalfirepower-scraper/ Then my dag script is refactored this way (so clean : I like it !) : import logging
import shutil
from datetime import datetime
from airflow.decorators import dag, task
log = logging.getLogger(__name__)
if not shutil.which("virtualenv"):
log.warning(
"This Dag requires virtualenv, please install it."
)
else:
@dag(dag_id= 'amazing_scraper', schedule_interval='@daily', start_date=datetime(2021, 1, 1), catchup=False, tags=['scrapper'])
def taskflow():
"""
### TaskFlow API example using virtualenv
"""
@task.virtualenv(
requirements=[
'globalfirepower-scraper==0.0.1',
],
python_version=3.9,
system_site_packages=True
)
def extract():
"""
#### Extract task
Scrape Global Fire Power Countries Ranking in 2022.
"""
from globalfirepower import GlobalFirePowerScraper
scraper = GlobalFirePowerScraper
return scraper.get_country_ranking_table()
countries_ranking = extract()
dag = taskflow()And ... it doesn't work, and I'm in pain finding why and how I can fix it. Things I've checked before coming here :
This is the output of my DAG : Any idea ? I have exactly the same output on Heroku where my Python version is 3.9 in both setup and VirtualEnvOperator. Sorry if it's a silly question, I'm really new at Airflow and Data Engineering but eager to learn ! Regards |
Beta Was this translation helpful? Give feedback.
-
|
Cool! Glad to hear! |
Beta Was this translation helpful? Give feedback.
-
|
@potiuk wouldn't it be great if we wanted to have virtual env for the whole dag? Besides helper codes to packaging modules seems like a few extra works for me. Any other work around there? Thank you so much for helping the community. Appreciated. |
Beta Was this translation helpful? Give feedback.
-
This is a bit more complex. For worker, having a venv per DAG is not really useful because each Task in DAG is executed potentially on a different machine and you would anyhow have to dynamically create/delete such an env. But we are going to have a new operator - PythonPreexisingVirtualenvOperator - #25780 (close to be merged for 2.4.0) that should handle the case where you can create an immutable venv upfront in your environment and use it to execute your tasks in it. When it comes to parsing the whole DAG - this is a different story and it should be handled much better with combinations of two AIPs:
|
Beta Was this translation helpful? Give feedback.
-
|
Thank you so much @potiuk. I am also working on airflow pretty recently. I will try to follow what you have suggested so far. Appreciated. |
Beta Was this translation helpful? Give feedback.
-
|
you should also be able to package up your local dependency into a whl file, store it somewhere on the airflow server and put the filepath in the requirements section of the virtualenv operator. then you it should pip install that whl file just like the other requirements and you can import and run it within the operator callable |
Beta Was this translation helpful? Give feedback.
-
|
We've added the whole dags folder to PYTHONPATH=/opt/airflow/dags# dag_do_some_etl.py
from acme_lib.salesforce_etl import utils_without_dependencies
...
@task.virtualenv(requirements=["some-required-dependency"])
def virtualenv_task():
from acme_lib.salesforce_etl import utils_with_dependencies
...BR |
Beta Was this translation helpful? Give feedback.
-
|
Not sure why you would need to do it, because dag on PYTHONPATH generally should be added automatically added by Airflow. (https://airflow.apache.org/docs/apache-airflow/stable/administration-and-deployment/modules_management.html#built-in-pythonpath-entries-in-airflow). Maybe the modified PYTHONPATH added by airflow is not propageted to a virtualenv operator ? That would be a nice feature to add if that's the case. And I don't see why it would be a problem - this is the recommended way we tell people to add and refer to shared utilities (see the link above). |
Beta Was this translation helpful? Give feedback.
-
|
Hey @potiuk Thanks for the evaluation! As I've understood it, Airflow adds those folders to It would indeed be a nice feature to optionally propagate the |
Beta Was this translation helpful? Give feedback.
-
|
Yeah. Maybe a good idea to add it as an option to Python Virtualenv Operator ? |
Beta Was this translation helpful? Give feedback.
-
|
Yes. The reason we use the path solution is that local changes (we use docker compose for our dev envs) are automatically reflected in the tasks this way (without having to re-build the package beforehand) Maybe adding a simple |
Beta Was this translation helpful? Give feedback.
-
|
Optionally when specified (for back compatibility) - yes. PRs are most welcome. |
Beta Was this translation helpful? Give feedback.
Package your
solvolabscode inpippackage and add it as requirement to virtualenv. Implement pipeline where you can release a new version of the code topipvia "single click".This is the only approach that is not a hack that makes sense I think.