Hey all! I am using airflow to orchestrate my scheduled meltano jobs, however it seems that environment variables set on schedules in meltano.yml are ignored. I pasted the relevant parts of my meltano.yml below. When I run the schedule using

meltano schedule run extract-load

, everything works as expected. I found that for elt schedules, airflow calls

meltano schedule run

, but for job schedules, airflow calls

meltano run

, ignoring all schedule-level settings. Is there any way to define which dbt models I want to run per schedule? Or am I using it wrong?

version: 1
plugins:
  extractors: [...]
  orchestrators: [...]
  transformers: [...]
  files: [...]
schedules:
  - name: extract-load
    interval: 0 0 * * *
    job: EL
    env:
      DBT_MODELS: calendar appdata woocommerce shopify
jobs:
  - name: EL
    tasks:
      - tap-calendar target-postgres
      - tap-appdata target-postgres
      - tap-woocommerce target-postgres
      - tap-shopify target-postgres
      - dbt:run
environments: [...]

That should work and if it doesn’t then you’ve found a bug. Can you share which version of Meltano you’re on? @florian.hines nothing about

run

should be ignoring a schedule env like that, right?

I'm on v2.4.0. I think the problem lies in the way jobs are executed in airflow. I took a look at the auto-generated

orchestrate/dags/meltano.py

and for job-type schedules, the command is not

meltano schedule run

which would respect env vars set on that level, but instead

meltano run <task>

@taylor I think that's expected behavior - it pop'd up during one of the discussions when we where adding support for scheduled jobs and was one of the reasons originally why I'd started to roll with still using

meltano schedule run

@florian.hines do we already have an issue on that? At face value it does break the contract of

env

on schedules in an unexpected way. I totally get why we have the current state though.

no issue on it but i'll go pop one. Its a super low weight change.

@florian.hines @taylor Thanks for your help, really appreciate the support! For now, overwriting the generated

orchestrate/dags/meltano.py

in a way that

meltano schedule run

is invoked should be a workaround, right?

@benjamin_mitzkus yep!

@benjamin_mitzkus if you're feeling adventurous, I think the change we'd be making in the operator is actually a pretty simple one. Probably just changing the BashOperator call to look like:

task = BashOperator(
    task_id=task_id,
    bash_command=f"cd {PROJECT_ROOT}; {MELTANO_BIN} run {run_args}",
    dag=dag,
    env=schedule.get("env", {}),
    append_env=True,
)

@florian.hines I'll take a look this evening 😃

Logged an issue - https://github.com/meltano/files-airflow/issues/32

@florian.hines do we need that on the extension as well?

yep - https://github.com/meltano/airflow-ext/issues/5#issuecomment-1252516415