-
Notifications
You must be signed in to change notification settings - Fork 175
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Running Dbt test command using DbtDag in Airflow #1330
Comments
Hey @sbutol01! I'm here to help you with your Cosmos issue. Let's tackle this together! To run the However, if you want to achieve this within the context of a DAG setup similar to from cosmos.operators.local import DbtTestLocalOperator
dbt_test_task = DbtTestLocalOperator(
task_id='dbt_test',
# Add any additional parameters you need here
) If you are set on using |
Thanks @dosu for your quick response. Actually I also used DbtTestLocalOperator, it does work also, but at the end it just fails my dag, giving me error like this - ERROR�[0m - Error during DBT run: Operator <Task(DbtTestLocalOperator): test> has not been assigned to a DAG yet�[0m I get this error after the test execution are successfully ran. Earlier I thought this will be a quick fix, but then unfortunately I couldn't get this fixed. Below is the skeleton code I'm using for this. Also please note that I'm doing all this within my virtualenv using @task.virtualenv. Below code does run the dag, but fails with above mentioned error, I tried modifying the code, but everytime I get error saying operator has not been assigned to dag yet. Could there be a quick fix. Thanks.
|
To resolve the error "Operator <Task(DbtTestLocalOperator): test> has not been assigned to a DAG yet," ensure that the In your code, the @dag(
dag_id=Path(__file__).stem,
start_date=pendulum.datetime(2024, 9, 12, tz="Pacific/Auckland"),
tags=["test-tag"],
max_active_runs=1,
catchup=False,
)
def dbt_test_process():
"""
The Airflow Dag to invoke dbt tests
"""
from cosmos import DbtTestLocalOperator, ProfileConfig
test_operator = DbtTestLocalOperator(
profile_config=profile_config,
project_dir=local_dir,
task_id="test",
dbt_cmd_flags=["--models", "model_name"],
append_env=True,
)
@task.virtualenv(
task_id="dbt_test_using_operator",
requirements=[
"dbt-snowflake==1.7.0",
"astronomer-cosmos==1.7.0",
],
)
def run_dbt_using_operators():
try:
logger.info("Executing DBT test operator...")
test_operator.execute(context={})
logger.info("DBT run completed successfully.")
except Exception as e:
logger.error(f"Error during DBT run: {e}")
raise
return "DBT task completed successfully"
run_dbt_using_operators()
dbt_test_process() Ensure that the |
To resolve the
By ensuring that |
Hey @sbutol01, How are you installing |
Hello experts,
I'm pretty new to Cosmos and trying to run my dbt commands using Cosmos in Airflow.
I'm trying to run dbt test command using Cosmos operators DbtDag. But to my surprise it doesn't run dbt test, instead its using the dbt run command. Is there a way to pass in an argument to force it to run the command I want it to run. I have below code snippet, I tried using operator_args, but looks like it doesn't accept dbt_command parameter.
dbt_dag = DbtDag(
project_config=ProjectConfig(local_dir),
profile_config=profile_config,
execution_config=execution_config,
operator_args={
"dbt_command": "test",
"dbt_args": ["--select", "model_name"],
},
dag_id="dbt_dag",
)
dbt_dag.test()
**gives me below error.
airflow.exceptions.AirflowException: Invalid arguments were passed to DbtRunLocalOperator (task_id: run). Invalid arguments were:
[2024-11-18, 15:43:54 ] {{process_utils.py:190}} INFO - kwargs: {'dbt_command': 'test', 'dbt_args': ['--select', 'model_name']}
Earlier I had used below code and it did run but it ran with below command. Looks like test() function isn't doing anything.
INFO - Trying to run dbtRunner with:
[2024-11-15, 13:55:44 ] {{process_utils.py:190}} INFO - ['run', '--models', 'model_name'.....]
dbt_dag = DbtDag(
project_config=ProjectConfig(local_dir),
profile_config=profile_config,
execution_config=execution_config,
dag_id="dbt_dag",
)
dbt_dag.test()
I know there is DbtTestLocalOperator, but I want to achieve test command execution using DbtDag.
Any help would be highly appreciated.
The text was updated successfully, but these errors were encountered: