fix: update outlets and schedule datasets trigger to ODS_USERDATA - #349
Conversation
There was a problem hiding this comment.
Pull request overview
This PR fixes an Airflow Dataset-trigger mismatch so the union_member_to_backend DAG is triggered by the dataset event produced by the ODS_USER_INFO.info DAG.
Changes:
- Updated
union_member_to_backendto listen toDataset("ODS_USERDATA")instead ofDataset("ODS_USERDATA.student_id"). - Updated
ODS_USER_INFO.infoproducer task to publishDataset("ODS_USERDATA")as an outlet.
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 3 comments.
| File | Description |
|---|---|
dags/ods/user/user_to_back.py |
Aligns the consumer DAG’s Dataset schedule with the producer’s published Dataset name. |
dags/ods/user/info.py |
Changes the producer task’s outlet Dataset to the unified name used by the consumer. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| # см. info.sql и https://airflow.apache.org/docs/apache-airflow/stable/templates-ref.html#variables | ||
| inlets=[Dataset("STG_USERDATA.info"), Dataset("STG_USERDATA.param"), Dataset("STG_UNION_MEMBER.union_member")], | ||
| outlets=[Dataset("ODS_USER_INFO.info")], | ||
| outlets=[Dataset("ODS_USERDATA")], |
There was a problem hiding this comment.
Dataset naming in this repo appears to consistently follow the SCHEMA.table pattern (e.g. ODS_AUTH.user, ODS_RATING.comment, etc.). Introducing a schema-level dataset name without a dot (ODS_USERDATA) is inconsistent and can make lineage/triggering harder to reason about. Consider publishing a table-scoped dataset (e.g. ODS_USERDATA.student_id) or publishing multiple outlets for the tables written in info.sql, and update downstream schedules accordingly.
| outlets=[Dataset("ODS_USERDATA")], | |
| outlets=[Dataset("ODS_USERDATA.info")], |
| # см. info.sql и https://airflow.apache.org/docs/apache-airflow/stable/templates-ref.html#variables | ||
| inlets=[Dataset("STG_USERDATA.info"), Dataset("STG_USERDATA.param"), Dataset("STG_UNION_MEMBER.union_member")], | ||
| outlets=[Dataset("ODS_USER_INFO.info")], | ||
| outlets=[Dataset("ODS_USERDATA")], |
There was a problem hiding this comment.
PR description says the producer (DAG ODS_USER_INFO.info) was already publishing ODS_USERDATA, but in this diff the producer outlet is being changed from ODS_USER_INFO.info to ODS_USERDATA. Please update the PR description to reflect the actual before/after state (or clarify why Airflow showed a different produced dataset than what the code had).
| with DAG( | ||
| dag_id="union_member_to_backend", | ||
| schedule=[Dataset("ODS_USERDATA.student_id")], | ||
| schedule=[Dataset("ODS_USERDATA")], |
There was a problem hiding this comment.
This DAG’s dataset schedule is now ODS_USERDATA (schema-level). Elsewhere in the repo, dataset-triggering is consistently done at table granularity (SCHEMA.table). To stay consistent and avoid introducing a one-off dataset naming scheme, consider switching back to a table-scoped dataset (e.g. ODS_USERDATA.student_id) and ensuring the producer publishes that same dataset.
| schedule=[Dataset("ODS_USERDATA")], | |
| schedule=[Dataset("ODS_USERDATA.student_id")], |
Уточню: он публиковал ODS_USER_INFO.info т.е. вообще несуществующую в бд схему. А должен был как раз юзердату |
| @@ -334,7 +334,7 @@ def remove_non_union_members_from_union_group(union_members_ids: list): | |||
|
|
|||
| with DAG( | |||
| dag_id="union_member_to_backend", | |||
There was a problem hiding this comment.
Хорошо бы в этом даге еще добавить inlets в явном виде, чтобы видеть зависимость на графе
There was a problem hiding this comment.
Возможно стоит поменять название дага в соответствии с целевой схемой - вопрос на подумать на самом деле
| # см. info.sql и https://airflow.apache.org/docs/apache-airflow/stable/templates-ref.html#variables | ||
| inlets=[Dataset("STG_USERDATA.info"), Dataset("STG_USERDATA.param"), Dataset("STG_UNION_MEMBER.union_member")], | ||
| outlets=[Dataset("ODS_USER_INFO.info")], | ||
| outlets=[Dataset("ODS_USERDATA")], |
There was a problem hiding this comment.
почему кстати тогда этот даг не падал, если у нас в dwh нет ODS_USER_INFO.info?
There was a problem hiding this comment.
inlets и outlets параметры по сути это просто наша договоренность. Airflow не проверяет какие реально схемы заполняются и т.п. поэтому это в наших интересах их писать так чтобы они соответствовали тому как действует даг, но если так не происходит то ничего не случается. Эирфлоу просто будет думать что такой датасет существет
… ods.user.info.py
|
💩 Code linting failed, use |
Проблема:
DAG
union_member_to_backendне запускался по Dataset-триггеру, так как producer (DAGODS_USER_INFO.info) и consumer использовали разные имена Dataset:ODS_USERDATAODS_USERDATA.student_idИз-за этого Airflow не создавал событие для ожидаемого Dataset, и DAG не триггерился.
Решение:
ODS_USERDATAODS_USERDATA