harrytandata/airflow_course

Sample code for Harry's Airflow online trainng course

You can find the videos on youtube or bilibili.

I am working on adding below things:

1. the slide pdf files(done)
2. another video about creating custom operators
3. docker-compose with CeleryExecutors.

airflow_course's Issues

[Question] `dbt deps`命令安排在哪个阶段

Harry 老师：

我在airflow中集成dbt的过程比较顺利，但在关于dbt deps命令的安排上有些困惑。
我的airflow部署在k8s中，采用deploy与dags分离的策略，无论是web, scheduler, 还是worder，都通过git将DAG仓库定时同步到容器中，dags目录挂载模式是emptyDir。与你的airflow_course类似，我的dags与dbt project在同一个仓库中，方便开发。

下面是我的自定义Operator代码片段：

@cached(cache=TTLCache(maxsize=1, ttl=60))
def get_finance_dbt_env_task()-> dict[str, str]:
    ...
class FinanceDbtOperator(BashOperator):
    def __init__(self, task_id: str, bash_command: str, **kwargs):
        _dbt_env = get_finance_dbt_env_task()
        kwargs['cwd'] = _dbt_env['DBT_PROJECT_DIR']
        kwargs['env'] = _dbt_env
        kwargs['append_env'] = True
        if 'queue' not in kwargs:
            kwargs['queue'] = 'kubernetes'
        super().__init__(task_id = task_id, bash_command = bash_command, **kwargs)

调用时仅需：

_dbt_task = FinanceDbtOperator(
    task_id='dbt_run',
    bash_command="dbt run --models 'dw.dim_stock'",
)

每运行一个task实例，都对应启动一个pod，拉取到干净的dbt项目文件后，都需要运行一次dbt deps以安装项目必要的包。
我实践或思考了一些方法：

方案	优点	缺点
在FinanceDbtOperator调整为'dbt deps\n' + bash_command	省事儿	每次运行任务都自动重复一遍，产生大量spin up阶段的消耗
将包含dags与dbt项目的dags目录持久化	资源消耗低	仍然难以判断应何时运行dbt deps（packages.yml是否发生更改）。反复运行dbt deps时，由于挂载了持久化卷，所以资源占用不大，但仍然需要花费数秒时间
将dbt deps移动到airflow deploy阶段	资源消耗低	dags仓库中的依赖项变更需要运维人员的干预，且难以应对多个dbt项目
为dbt项目构建独立的镜像	运行时资源消耗低，且能够应对多个dbt项目的依赖冲突	需要构建复杂的CI/CD管道，让airflow中的DAG与dbt项目的镜像协同运作
将dbt_packages从.gitignore中移除，所有依赖包都纳入代码仓库	资源消耗低，容易协同	很脏的一种做法

想要了解，在你们的airflow与dbt集成的实践中，是如何安排dbt deps的执行时机的呢？