Skip to content

fix: update outlets and schedule datasets trigger to ODS_USERDATA #349

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.

Sign up for GitHub

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

Merged
petrCher merged 3 commits into from
Apr 25, 2026
Merged

fix: update outlets and schedule datasets trigger to ODS_USERDATA #349

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
2c2404e
fix: update outlets and schedule datasets trigger to ODS_USERDATA
sanycool Apr 24, 2026
c731854
add inlets and fix logging in ods.user.user_to_back, change DAG id in…
sanycool Apr 25, 2026
20c4198
fix lint
sanycool Apr 25, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions dags/ods/user/info.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@


with DAG(
dag_id="ODS_USER_INFO.info",
dag_id="ODS_USERDATA",
start_date=datetime(2024, 10, 1),
schedule=[Dataset("STG_USERDATA.info"), Dataset("STG_USERDATA.param"), Dataset("STG_UNION_MEMBER.union_member")],
catchup=False,
Expand All @@ -31,7 +31,7 @@
# порядок датасетов здесь важен!
# см. 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")],

Copilot AI Apr 24, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Suggested change
outlets=[Dataset("ODS_USERDATA")],
outlets=[Dataset("ODS_USERDATA.info")],

Copilot uses AI. Check for mistakes.

Copilot AI Apr 24, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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).

Copilot uses AI. Check for mistakes.

@petrCher petrCher Apr 24, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Image почему кстати тогда этот даг не падал, если у нас в dwh нет ODS_USER_INFO.info?

@petrCher petrCher Apr 24, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@RedstoneEnjoyer подскажи если знаешь

@RedstoneEnjoyer RedstoneEnjoyer Apr 24, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

inlets и outlets параметры по сути это просто наша договоренность. Airflow не проверяет какие реально схемы заполняются и т.п. поэтому это в наших интересах их писать так чтобы они соответствовали тому как действует даг, но если так не происходит то ничего не случается. Эирфлоу просто будет думать что такой датасет существет

@petrCher petrCher Apr 24, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ок, понял, спасибо

@petrCher petrCher Apr 24, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

тогда ждем добавления inlets от @sanycool

params={"tablename": "info"},
)
# run_sql_encrypted = PostgresOperator(
Expand All @@ -46,6 +46,6 @@
# Dataset("STG_USERDATA.param"),
# Dataset("STG_UNION_MEMBER.union_member"),
# ],
# outlets=[Dataset("ODS_USER_INFO.encrypted_info")],
# outlets=[Dataset("ODS_USERDATA.encrypted_info")],
# params={"tablename": "encrypted_info"},
# ) TODO сделать мерж юсердаты шифрованной
7 changes: 4 additions & 3 deletions dags/ods/user/user_to_back.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -273,10 +273,10 @@ def post_members_to_union_group_to_backend(union_members_ids: list):
json=data,
)
if response.status_code == 200:
logging.info(f"Updated group with union members {union_members_ids} from dwh database")
logging.info(f"Updated group with union member {union_member} from dwh database")
else:
logging.error(
f"Update group with union members {union_members_ids} copy to backend failed with code: {response.status_code}\n Response text: {response.text}"
f"Update group with union member {union_member} copy to backend failed with code: {response.status_code}\n Response text: {response.text}"
)
except Exception as e:
logging.error(f"Error sending data to backend: {str(e)}")
Expand Down Expand Up @@ -334,7 +334,8 @@ def remove_non_union_members_from_union_group(union_members_ids: list):

with DAG(
dag_id="union_member_to_backend",

@RedstoneEnjoyer RedstoneEnjoyer Apr 24, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Хорошо бы в этом даге еще добавить inlets в явном виде, чтобы видеть зависимость на графе

schedule=[Dataset("ODS_USERDATA.student_id")],
schedule=[Dataset("ODS_USERDATA")],

Copilot AI Apr 24, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Suggested change
schedule=[Dataset("ODS_USERDATA")],
schedule=[Dataset("ODS_USERDATA.student_id")],

Copilot uses AI. Check for mistakes.
inlets=[Dataset("ODS_USERDATA")],
start_date=datetime.datetime(start_year, start_month, start_day),
catchup=False,
tags=["ods", "userdata", "union_member", "backend"],
Expand Down
Loading