Elementary is the open-source data observability platform built as a dbt package. Add to packages.yml: - package: elementary-data/elementary, version: 0.14.0. dbt deps && dbt run --select elementary sets up Elementary tables. Elementary tests in schema.yml: tests: - elementary.volume_anomalies: {time_bucket: {period: day}, days_back: 14}, - elementary.all_columns_anomalies, - elementary.freshness_anomalies: {max_allowed_delay: 6 hours}. Column-level anomalies: - elementary.column_anomalies: {column_anomalies: [null_rate, zero_rate, min, max, average]}. Schema changes: - elementary.schema_changes. Distribution: - elementary.event_freshness_anomalies: {event_timestamp_column: event_time}. edr report generates an interactive HTML report. edr send-report --slack-token TOKEN --slack-channel alerts sends to Slack. edr monitor runs tests and alerts. Elementary CLI: pip install elementary-data[bigquery]. Source freshness: - elementary.source_freshness_anomalies. edr run-report in CI/CD. Custom monitors via macros: {{ elementary.collect_metrics(model, metrics=['row_count', 'null_rate']) }}. Python SDK: from elementary.clients.dbt.api import DbtRunner. Elementary Cloud at elementary-data.cloud stores history, provides root cause analysis, and routes alerts by team. edr debug diagnoses connection issues. profiles.yml database connection mirrors dbt profile. Claude Code generates Elementary dbt test YAML, report configurations, Slack alert setups, and CI/CD integration workflows.
CLAUDE.md for Elementary
## Elementary Stack
- Package: elementary-data/elementary >= 0.14 in packages.yml
- Install: dbt deps && dbt run --select elementary (creates edr_monitors_runs and artifact tables)
- Tests: add to schema.yml under tests: - elementary.volume_anomalies / all_columns_anomalies / etc.
- Run: dbt test --select elementary (runs all elementary tests)
- Report: edr report (HTML) or edr send-report --slack-token / --slack-channel
- CI: dbt test && edr send-report on failure
- Cloud: edr report --cloud-report (upload to Elementary Cloud)Elementary Test Configuration
# models/marts/schema.yml — Elementary observability tests
version: 2
models:
- name: orders_daily
description: "Daily enriched orders"
config:
elementary:
timestamp_column: created_at # used for time-series monitoring
tests:
# Volume monitoring — detect sudden drops/spikes
- elementary.volume_anomalies:
timestamp_column: created_at
days_back: 30
time_bucket:
period: hour
sensitivity: 3 # standard deviations
anomaly_direction: both
# Schema change detection
- elementary.schema_changes
# Column-level anomalies across all columns
- elementary.all_columns_anomalies:
timestamp_column: created_at
days_back: 14
time_bucket: { period: day }
column_anomalies:
- null_rate
- null_count
- zero_rate
- average
- variance
# Freshness (data recency)
- elementary.freshness_anomalies:
max_allowed_delay: "6 hours"
days_back: 14
columns:
- name: order_id
tests:
- not_null
- unique
- elementary.column_anomalies:
column_anomalies: [null_count]
- name: amount_usd
tests:
- elementary.column_anomalies:
column_anomalies: [null_rate, zero_rate, min, max, average, variance]
timestamp_column: created_at
days_back: 30
- name: status
tests:
- accepted_values:
values: [pending, processing, completed, refunded, cancelled]
- elementary.column_anomalies:
column_anomalies: [null_rate]
- name: events
config:
elementary:
timestamp_column: event_timestamp
tests:
- elementary.volume_anomalies:
timestamp_column: event_timestamp
time_bucket: { period: hour }
days_back: 7
anomaly_direction: drop # Alert only on drops (events never spike unexpectedly)
sensitivity: 2
- elementary.event_freshness_anomalies:
event_timestamp_column: event_timestamp
max_allowed_delay: "30 minutes"
- elementary.all_columns_anomalies:
timestamp_column: event_timestamp
days_back: 14
sources:
- name: raw
schema: raw_data
tables:
- name: api_events
tests:
- elementary.source_freshness_anomalies:
timestamp_column: _fivetran_synced
max_allowed_delay: "2 hours"Elementary Package Config
# packages.yml — dbt packages
packages:
- package: dbt-labs/dbt_utils
version: 1.1.1
- package: elementary-data/elementary
version: 0.14.3# dbt_project.yml — Elementary configuration block
vars:
# Elementary config
edr_schema: elementary # Schema where Elementary tables are stored
edr_database: analytics # Database for Elementary tables (some warehouses only)
# Enable/disable
elementary_enabled: true # Set false in dev to skip anomaly tests
anomaly_sensitivity: 3 # Standard deviations for anomaly detection
# Date range for anomaly training
rolling_lookback: 30 # Days of history to train on
days_back: 14 # Days to check for anomaliesCI/CD Integration
#!/bin/bash
# scripts/dbt_ci_with_elementary.sh — CI/CD pipeline with Elementary monitoring
set -e
echo "=== Running dbt build ==="
dbt build --target prod --exclude tag:skip_ci
echo "=== Running Elementary tests ==="
dbt test --select elementary --target prod
echo "=== Generating Elementary report ==="
edr report --target prod --env prod
# Send report to Slack on test failures
if [ $? -ne 0 ]; then
echo "=== Sending failure report to Slack ==="
edr send-report \
--target prod \
--slack-token "$SLACK_BOT_TOKEN" \
--slack-channel "#data-alerts" \
--env prod
exit 1
fi
echo "=== Elementary monitoring complete ==="# scripts/elementary_alert.py — programmatic Elementary alerting
import os
import json
from pathlib import Path
def parse_elementary_results(results_path: str = "target/run_results.json") -> list[dict]:
"""Parse dbt run results for Elementary failures."""
with open(results_path) as f:
results = json.load(f)
failures = []
for r in results.get("results", []):
if r.get("status") in ("fail", "warn") and "elementary" in r.get("unique_id", ""):
failures.append({
"test": r["unique_id"],
"status": r["status"],
"message": r.get("message", ""),
"node": r.get("node", {}).get("name", ""),
"model": r.get("node", {}).get("relation_name", ""),
"test_type": r.get("node", {}).get("test_metadata", {}).get("name", ""),
})
return failures
def send_slack_alert(failures: list[dict], webhook_url: str) -> None:
"""Send Slack alert for Elementary failures."""
import urllib.request
if not failures:
return
blocks = [
{
"type": "header",
"text": {"type": "plain_text", "text": f"⚠️ {len(failures)} Data Quality Alert(s)"},
},
]
for f in failures[:10]: # Limit to 10 in Slack message
blocks.append({
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"*{f['test_type']}* on `{f['node']}`\n{f['message'][:200]}",
},
})
payload = json.dumps({"blocks": blocks}).encode()
req = urllib.request.Request(
webhook_url,
data=payload,
headers={"Content-Type": "application/json"},
method="POST",
)
urllib.request.urlopen(req)
if __name__ == "__main__":
failures = parse_elementary_results()
if failures:
print(f"Found {len(failures)} Elementary test failures")
webhook = os.environ.get("SLACK_WEBHOOK_URL")
if webhook:
send_slack_alert(failures, webhook)
import sys; sys.exit(1)
print("All Elementary tests passed")For the Soda alternative when needing a standalone data quality tool that works without dbt — Soda Core connects directly to any warehouse and runs SodaCL checks without requiring a dbt project, making it useful for data pipelines not built on dbt while Elementary lives entirely inside the dbt ecosystem. For the Monte Carlo alternative when needing a fully managed enterprise data observability platform with automatic anomaly detection, data lineage, SAML SSO, and SLA-backed incident management without any configuration — Monte Carlo deploys as SaaS with a UI while Elementary is open-source and self-configurable, requiring more upfront setup in exchange for no vendor lock-in. The Claude Skills 360 bundle includes Elementary skill sets covering dbt test YAML, anomaly detection configuration, CI/CD integration, and Slack alerting. Start with the free tier to try data observability generation.