Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
109e009210 | ||
|
|
13df1026cd |
+5
-6
@@ -1,9 +1,8 @@
|
||||
# dbt build artefacts
|
||||
target/
|
||||
dbt_modules/
|
||||
dbt_packages/
|
||||
logs/
|
||||
# Keep package model sources under models/logs/ (not the dbt logs/ dir)
|
||||
# dbt build artefacts (root-only — do NOT ignore models/logs/)
|
||||
/target/
|
||||
/dbt_modules/
|
||||
/dbt_packages/
|
||||
/logs/
|
||||
|
||||
# Local / secrets
|
||||
.user.yml
|
||||
|
||||
@@ -4,7 +4,7 @@ Modernized fork of [re_data](https://github.com/re-data/re-data) for fast2.dev D
|
||||
|
||||
- **dbt package name:** `f2dq` (what `dbt_project.yml` / `detect_namespace()` use)
|
||||
- **Hub-style identity:** `dbt_f2dq`
|
||||
- **Version:** `1.0.0`
|
||||
- **Version:** `1.0.1`
|
||||
- **Requires:** dbt-core `>=1.0,<2.0`, `dbt_utils==1.4.1`
|
||||
|
||||
## Install (git — recommended)
|
||||
@@ -12,7 +12,7 @@ Modernized fork of [re_data](https://github.com/re-data/re-data) for fast2.dev D
|
||||
```yaml
|
||||
packages:
|
||||
- git: "https://gitea.m7xlab.top/fast2dev/dbt-f2dq.git"
|
||||
revision: v1.0.0
|
||||
revision: v1.0.1
|
||||
```
|
||||
|
||||
Then:
|
||||
@@ -23,6 +23,10 @@ dbt deps
|
||||
|
||||
Pin a release tag in production; use `main` only if you want floating tip.
|
||||
|
||||
Default warehouse schemas: `f2dq` (public models) and `f2dq_internal`
|
||||
(internal models). Override in the consuming project's `dbt_project.yml`
|
||||
under `models.f2dq`.
|
||||
|
||||
## Install (local path — active development)
|
||||
|
||||
Prefer a local path while editing the package itself:
|
||||
@@ -34,12 +38,22 @@ packages:
|
||||
|
||||
## Dual-read (migration)
|
||||
|
||||
Node config keys accept either prefix:
|
||||
Node config keys accept either prefix (config top-level **or** `config.meta` /
|
||||
node `meta` — prefer meta on dbt 1.12+ to avoid CustomKeyInConfigDeprecation):
|
||||
|
||||
- `+f2dq_monitored` / `+re_data_monitored`
|
||||
- `+f2dq_time_filter` / `+re_data_time_filter`
|
||||
- `f2dq_monitored` / `re_data_monitored`
|
||||
- `f2dq_time_filter` / `re_data_time_filter`
|
||||
- (same for metrics, columns, anomaly_detector, owners)
|
||||
|
||||
```yaml
|
||||
models:
|
||||
my_project:
|
||||
staging:
|
||||
+meta:
|
||||
f2dq_monitored: true
|
||||
f2dq_time_filter: created_at
|
||||
```
|
||||
|
||||
Vars: `f2dq:<name>` preferred; boolean enable flags also honour `re_data:<name>`
|
||||
(`store_table_samples`, `save_test_history`, etc.) via `f2dq.f2dq_var()`.
|
||||
|
||||
|
||||
+19
-5
@@ -1,5 +1,5 @@
|
||||
name: "f2dq"
|
||||
version: "1.0.0" # dbt_f2dq: modernized fork (dbt_utils 1.4.1); dual-read re_data_* keys
|
||||
version: "1.0.2" # dbt_f2dq: modernized fork (dbt_utils 1.4.1); dual-read re_data_* keys
|
||||
config-version: 2
|
||||
|
||||
require-dbt-version: [">=1.0.0", "<2.0.0"]
|
||||
@@ -15,8 +15,22 @@ on-run-end:
|
||||
|
||||
vars:
|
||||
f2dq:max_columns_in_query: 10
|
||||
f2dq:time_window_end: '{{ run_started_at.strftime("%Y-%m-%d 00:00:00") }}'
|
||||
f2dq:time_window_start: '{{ (run_started_at - modules.datetime.timedelta(1)).strftime("%Y-%m-%d 00:00:00") }}'
|
||||
# The metric bucket is the run's calendar day: [midnight of the run date, +1d).
|
||||
#
|
||||
# re_data used [run_date - 1d, run_date) - i.e. the bucket ENDED at midnight at
|
||||
# the start of the run day. Any row stamped during the run (models that set
|
||||
# `dbt_updated_at = CURRENT_DATETIME()`) therefore fell outside the bucket the
|
||||
# run had just written it into, and incremental merges that re-stamp rows every
|
||||
# run made those rows land in the wrong bucket or never. row_count came out 0/1
|
||||
# against tables that plainly had rows. Anchoring the bucket on the run date
|
||||
# instead of the day before keeps `interval_length_sec` at 86400 and the daily
|
||||
# series intact, but includes everything written on the day being measured.
|
||||
f2dq:time_window_start: '{{ run_started_at.strftime("%Y-%m-%d 00:00:00") }}'
|
||||
f2dq:time_window_end: '{{ (run_started_at + modules.datetime.timedelta(1)).strftime("%Y-%m-%d 00:00:00") }}'
|
||||
# Also emit `global__<metric>` (whole-table) alongside the windowed series, so
|
||||
# metrics stay meaningful for models whose time_filter holds historical
|
||||
# timestamps that miss the current bucket.
|
||||
f2dq:emit_global_metrics: true
|
||||
f2dq:anomaly_detection_look_back_days: 30
|
||||
f2dq:select: null
|
||||
f2dq:anomalies_filtered: f2dq_anomalies
|
||||
@@ -68,6 +82,6 @@ vars:
|
||||
|
||||
models:
|
||||
f2dq:
|
||||
+schema: f2
|
||||
+schema: f2dq
|
||||
internal:
|
||||
+schema: f2_internal
|
||||
+schema: f2dq_internal
|
||||
|
||||
@@ -19,6 +19,17 @@ sources:
|
||||
|
||||
vars:
|
||||
f2dq:store_table_samples: true
|
||||
# The expected_*.csv seeds encode re_data's original bucket anchoring
|
||||
# ([run_date - 1d, run_date)) and only the rows that bucket produces. The
|
||||
# package default now anchors the bucket on the run date and also emits
|
||||
# `global__*` metrics; both change the result set, and the seed values cannot
|
||||
# be regenerated without the upstream multi-database harness. Pin the legacy
|
||||
# behaviour here so this suite keeps exercising the metric engine and the
|
||||
# `global__`-vs-windowed naming contract. The corrected behaviour is covered
|
||||
# by the contract tests in the control_plane test suite.
|
||||
f2dq:emit_global_metrics: false
|
||||
f2dq:time_window_end: '{{ run_started_at.strftime("%Y-%m-%d 00:00:00") }}'
|
||||
f2dq:time_window_start: '{{ (run_started_at - modules.datetime.timedelta(1)).strftime("%Y-%m-%d 00:00:00") }}'
|
||||
f2dq:anomaly_detector:
|
||||
name: modified_z_score
|
||||
threshold: 0.6
|
||||
|
||||
@@ -37,37 +37,77 @@
|
||||
{% endmacro %}
|
||||
|
||||
{% macro metrics_base_insert(model, ref_model, columns, table_level=False) %}
|
||||
|
||||
{% set col_exprs = f2dq.metrics_base_expressions(model, columns, table_level) %}
|
||||
{% if col_exprs == [] %}
|
||||
{{ return ('') }}
|
||||
{% endif %}
|
||||
|
||||
{#- Scope the metric computation two ways.
|
||||
|
||||
'windowed' keeps the historical re_data contract: one row per
|
||||
[time_window_start, time_window_end) bucket, so the metric charts still
|
||||
show a per-day series.
|
||||
|
||||
'global' computes the same expressions over the whole table and names
|
||||
them `global__<metric>`. Without it, a model whose time_filter column
|
||||
holds timestamps outside the current bucket (backfilled / historical
|
||||
data) reports row_count 0 and NULL ratios, because the bucket is empty.
|
||||
`global__row_count` then describes the table as it actually is.
|
||||
|
||||
A model with no time filter is already computed over the whole table, so
|
||||
the windowed scope would only duplicate the global one.
|
||||
|
||||
Set `f2dq:emit_global_metrics` to false to restore windowed-only. -#}
|
||||
{% set emit_global = f2dq.f2dq_var('emit_global_metrics', true) %}
|
||||
{% if model.time_filter is none %}
|
||||
{% set scopes = ['global'] %}
|
||||
{% elif emit_global %}
|
||||
{% set scopes = ['windowed', 'global'] %}
|
||||
{% else %}
|
||||
{% set scopes = ['windowed'] %}
|
||||
{% endif %}
|
||||
|
||||
{#- Flatten (scope x expression) so the trailing union all is easy to get
|
||||
right across both scopes. -#}
|
||||
{% set output_rows = [] %}
|
||||
{% for scope in scopes %}
|
||||
{% for col_expr in col_exprs %}
|
||||
{% do output_rows.append({
|
||||
'scope': scope,
|
||||
'col_name': col_expr.col_name,
|
||||
'metric': col_expr.metric,
|
||||
'expr': col_expr.expr,
|
||||
}) %}
|
||||
{% endfor %}
|
||||
{% endfor %}
|
||||
|
||||
insert into {{ ref(ref_model) }}
|
||||
with temp_table_metrics as (
|
||||
with
|
||||
{%- for scope in scopes %}
|
||||
temp_table_metrics_{{ scope }} as (
|
||||
select
|
||||
{%- for col_expr in col_exprs %}
|
||||
( {{ col_expr.expr }} ) as {{ f2dq.quote_column_name(col_expr.col_name + '___' + col_expr.metric) }}
|
||||
{%- if not loop.last %},{%- endif %}
|
||||
{% endfor %}
|
||||
from
|
||||
{{ model.table_name }}
|
||||
where
|
||||
{{ in_time_window(model.time_filter) }}
|
||||
)
|
||||
|
||||
{%- for col_expr in col_exprs %}
|
||||
{% set final_metric_name = get_final_metric_name(col_expr.metric, model.time_filter) %}
|
||||
|
||||
select '{{model.table_name}}' as table_name, '{{ col_expr.col_name }}' as column_name, '{{ final_metric_name }}' as metric, {{ f2dq.quote_column_name(col_expr.col_name + '___' + col_expr.metric) }} as value
|
||||
from temp_table_metrics
|
||||
{% if not loop.last %}union all{% endif %}
|
||||
{% endfor %}
|
||||
{%- endfor %}
|
||||
from {{ model.table_name }}
|
||||
where {% if scope == 'global' %}true{% else %}{{ in_time_window(model.time_filter) }}{% endif %}
|
||||
){% if not loop.last %},{% endif %}
|
||||
{%- endfor %}
|
||||
|
||||
{%- for row in output_rows %}
|
||||
select
|
||||
'{{ model.table_name }}' as table_name,
|
||||
'{{ row.col_name }}' as column_name,
|
||||
'{{ f2dq.get_final_metric_name(row.metric, model.time_filter, row.scope) }}' as metric,
|
||||
{{ f2dq.quote_column_name(row.col_name + '___' + row.metric) }} as value
|
||||
from temp_table_metrics_{{ row.scope }}
|
||||
{%- if not loop.last %} union all{%- endif %}
|
||||
{%- endfor %}
|
||||
{% endmacro %}
|
||||
|
||||
{% macro get_final_metric_name(metric_name, time_filter) %}
|
||||
{% if time_filter is none %}
|
||||
{% macro get_final_metric_name(metric_name, time_filter, scope='windowed') %}
|
||||
{% if scope == 'global' or time_filter is none %}
|
||||
{{ return ('global__' + metric_name) }}
|
||||
{% else %}
|
||||
{{ return (metric_name) }}
|
||||
|
||||
@@ -36,7 +36,7 @@
|
||||
(
|
||||
select
|
||||
{{ overview_select_base('metric', 'computed_on')}}
|
||||
{{ to_single_json(['metric', 'value', 'time_window_end', 'interval_length_sec']) }} as {{ f2dq.quote_column('data') }}
|
||||
{{ to_single_json(['metric', 'value', 'time_window_start', 'time_window_end', 'interval_length_sec']) }} as {{ f2dq.quote_column('data') }}
|
||||
from
|
||||
{{ ref('f2dq_base_metrics') }}
|
||||
where {{ in_date_window('time_window_end', start_date, end_date) }}
|
||||
|
||||
@@ -9,6 +9,18 @@
|
||||
-- depends_on: {{ ref('f2dq_test_history') }}
|
||||
{% set command = flags.WHICH %}
|
||||
{% if execute and results and command in ('test', 'build') %}
|
||||
{% set history_rel = ref('f2dq_test_history') %}
|
||||
{% set history_exists = adapter.get_relation(
|
||||
database=history_rel.database,
|
||||
schema=history_rel.schema,
|
||||
identifier=history_rel.identifier
|
||||
) is not none %}
|
||||
{% if not history_exists %}
|
||||
{% do log(
|
||||
"f2dq_test_history relation missing (" ~ history_rel ~ "); skipping save_test_history",
|
||||
info=True
|
||||
) %}
|
||||
{% else %}
|
||||
{% set tests = [] %}
|
||||
{% for el in results %}
|
||||
{% if el.node.resource_type.value == 'test' %}
|
||||
@@ -18,12 +30,13 @@
|
||||
|
||||
{% if tests %}
|
||||
{% do f2dq.insert_list_to_table(
|
||||
ref('f2dq_test_history'),
|
||||
history_rel,
|
||||
tests,
|
||||
['table_name', 'column_name', 'test_name', 'status', 'execution_time', 'message', 'failures_count', 'failures_json', 'failures_table', 'severity', 'compiled_sql', 'run_at'],
|
||||
{ 'run_at': timestamp_type() }
|
||||
) %}
|
||||
{% endif %}
|
||||
{% endif %}
|
||||
|
||||
{% endif %}
|
||||
{{ return ('') }}
|
||||
|
||||
@@ -0,0 +1,23 @@
|
||||
{{
|
||||
config(
|
||||
materialized='incremental',
|
||||
on_schema_change='sync_all_columns',
|
||||
)
|
||||
}}
|
||||
|
||||
{{
|
||||
f2dq.empty_table_generic([
|
||||
('table_name', 'string'),
|
||||
('column_name', 'string'),
|
||||
('test_name', 'string'),
|
||||
('status', 'string'),
|
||||
('execution_time', 'numeric'),
|
||||
('message', 'string'),
|
||||
('failures_count', 'numeric'),
|
||||
('failures_json', 'long_string'),
|
||||
('failures_table', 'long_string'),
|
||||
('severity', 'string'),
|
||||
('compiled_sql', 'long_string'),
|
||||
('run_at', 'timestamp')
|
||||
])
|
||||
}}
|
||||
Reference in New Issue
Block a user