https://linen.dev logo
Join Slack
Powered by
Hey guys :slightly_smiling_face: I would like to a...
# community-support
j
Hey guys 🙂 I would like to ask if it's possible to have a generic test configuration in ex. volume_anomaly in dbt_project.yml which will run on all models in a specific schema? It works when I separately configure it for multiple models but I was thinking if there is a way to reuse the configuration and apply it on all the existing models in a specific schema?
j
Hey @Jelena! You might be able to do something with a macro, but I don't have any examples. ChatGPT gives me this as an option:
Copy code
{% macro apply_volume_anomaly_to_schema(schema_name) %}
  {% for model in graph.nodes.values() %}
    {% if model.schema == schema_name %}
      {% do model.config.update({"tests": [{"volume_anomaly": {"threshold": 0.1}}]}) %}
    {% endif %}
  {% endfor %}
{% endmacro %}
Again, this is not something I have tested, but it might be inspiration. You would have to change it make it elementary specific of course. In our cloud version we have the ability to add anomaly monitors on all of your assets, but in the open source this is not possible without specific configuration. Hope it helps!
j
Thanks @Joost Boonzajer Flaes this is very helpful as a start. Will look more into this
Hey @Joost Boonzajer Flaes I managed to do something with a macro. I will post it here so maybe someone will also find it useful
drop corrupt happy 1
Copy code
{% macro generate_schema_yml(schema_name, include_tags=[]) %}
  {% set models_in_schema = [] %}

  -- Collect models in the specific schema that also contain the specified tags
  {% for model in graph.nodes.values() %}
    {% if schema_name in model.schema
       and model.resource_type == "model"
       and (include_tags | length == 0 or model.tags | select("in", include_tags) | list | length > 0) %}
      {% do models_in_schema.append(model.name) %}
    {% endif %}
  {% endfor %}

  {% if models_in_schema | length == 0 %}
    {{ print("version: 2\n\nmodels: []") }}
    {{ return() }}
  {% endif %}

  {% set schema_yaml %}
version: 2

models:
{% for model in models_in_schema %}
  - name: {{ model }}
    description: "Automatically generated test for {{ model }}"
    tests:
      - elementary.volume_anomalies:
          fail_on_zero: true #If the table has zero rows for a time bucket (day), the test will fail.
          time_bucket:
            period: day
            count: 1 #Groups data into daily buckets
          detection_period:
            period: day
            count: 1 #Detects anomalies in the most recent full day - YESTERDAY
          training_period:
            period: day
            count: 4 # Learns from the last 4 days
         # optional - use tags to run elementary tests on a dedicated run
          tags: [ "bronze", "volume_anomaly" ]
          config:
            severity: warn
            elementary:
              timestamp_column: "filewindow"

      - elementary.freshness_anomalies:
          timestamp_column: "filewindow"
          detection_period:
            period: day
            count: 1  # Detect anomalies in the last full day
          training_period:
            period: day
            count: 4  # Learn from the last 4 days
          tags: [ "bronze","freshness_anomaly" ]
          config:
            severity: warn
{% endfor %}
  {% endset %}

  {% do log(schema_yaml, info=True) %}
  {{ print(schema_yaml) }}

{% endmacro %}
Copy code
dbt run-operation generate_schema_yml --args '{"schema_name": "bronze", "include_tags" :[" "]}' --quiet  | sed $'s/\x1b\[[0-9;]*m//g' > models/bronze/br_anomaly_tests.yml
🙌 1
What the macro is doing is going over the models based on the schema parameter and creates specific elementary tests. I have also added a tag parameter to potentially create different tests for different models
j
Very cool! Thanks for sharing the result 🙏
🙌 1
34 Views
Open in Slack
PreviousNext
Powered by