# Impulse Reporting

> Build and run an Impulse reporting pipeline that persists events and aggregations to the gold-layer star schema. Use when the user wants to "build an Impulse report", compute histograms/statistics across all recordings and write them to Delta tables for a dashboard or scheduled job, or run incrementally so only new/changed data is reprocessed. Covers the Report/Page lifecycle (`add_event`, `add_page`, `determine_report`, `persist_results`), incremental processing, and sinkless mode.

- Skill: `databrickslabs/impulse-reporting` (Agent Skill)
- Install (CLI): `npx skillmds@latest add databrickslabs/impulse-reporting`
- Raw SKILL.md: https://api.skillmd.com/api/skills/databrickslabs/impulse-reporting/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: DevOps & Infra
- Author: databrickslabs (https://skillmd.com/u/databrickslabs)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/databrickslabs/impulse-reporting

---


# Impulse — reporting pipeline

The reporting mode computes events and aggregations across all matching recordings and persists them
to the gold-layer star schema (see `impulse-data-model`), ready for AI/BI dashboards or Lakehouse apps.
`Report` (from `impulse_reporting.core.report`) is the orchestrator.

This skill covers the lifecycle. The pieces plug in from sibling skills: **`impulse-config`** (the
config dict), **`impulse-tsal`** (channel selection and signals), **`impulse-events`** (event windows),
and **`impulse-aggregations`** (histograms and statistics).

## Creating a Report

```python
from databricks.sdk import WorkspaceClient
from impulse_reporting.core.report import Report

ws = WorkspaceClient()

# From a dict
report = Report(name="my_report", spark=spark, workspace_client=ws, config=config_dict)

# From a JSON file
report = Report(name="my_report", spark=spark, workspace_client=ws, config_path="./config/config.json")
```

| Parameter          | Type              | Description                                            |
|--------------------|-------------------|--------------------------------------------------------|
| `name`             | `str`             | Report name; used to derive a unique `report_id`.      |
| `spark`            | `SparkSession`    | Active Spark session (present in Databricks notebooks). |
| `workspace_client` | `WorkspaceClient` | Authenticated Databricks SDK client.                   |
| `config`           | `dict`            | Config dict (or use `config_path`).                    |
| `config_path`      | `str`             | Path to a JSON config file (or use `config`).          |

Exactly one of `config` / `config_path` must be provided.

Useful methods: `get_db()` → the `MeasurementDB` for signal selection; `get_solver()` → the configured
solver; `get_sink_config()` → the resolved sink; `add_event(event)`; `add_page(page)`;
`add_calculated_channel(channel)` (see `impulse-channels`); `determine_report(is_incremental=None)`;
`persist_results(cleanup_temp_tables=None)` (pass `True`/`False` to override the
`unity_sink.cleanup_temp_tables` config flag for dropping `__impulse_temp_*` tables after a
successful write).

## The lifecycle

```python
from impulse_reporting.core.page import Page
from impulse_reporting.events.basic_event import BasicEvent
from impulse_reporting.aggregations.histogram import HistogramDuration

# 1. Select channels and derive signals (impulse-tsal)
db = report.get_db()
eng_rpm = db.query.channel(channel_name="Engine RPM", brand="Seat", model="Leon")
veh_spd = db.query.channel(channel_name="Vehicle Speed Sensor")

# 2. Define events and REGISTER them (impulse-events)
rpm_band = BasicEvent(name="rpm_band", expr=(eng_rpm > 2000) & (eng_rpm < 5000))
report.add_event(rpm_band)

# 3. Add a page and its aggregations (impulse-aggregations)
page = Page(page_number=1)
report.add_page(page)
page.add_aggregation(HistogramDuration(
    name="rpm_hist", base_expr=eng_rpm,
    bins=[0, 1000, 2000, 3000, 4000, 5000], event=rpm_band,
    channel_name="Engine RPM", bins_unit="rpm", values_unit="s",
))

# 4. (optional) Materialize derived channels (impulse-channels)
from impulse_reporting.channels.calculated_channel import CalculatedChannel
report.add_calculated_channel(CalculatedChannel(
    name="speed_kmh", expr=veh_spd * 3.6,
    identity={"channel_name": "speed_kmh", "data_key": "CALC"},
))

# 5. Compute everything (in parallel across containers)
report.determine_report()

# 6. Write the gold star schema
report.persist_results()
```

`determine_report()` validates that **every event referenced by an aggregation has been registered
with `add_event()`** before computing — otherwise it raises. Register events even when an aggregation
is the only thing that uses them.

After `determine_report()`, computed results are also available on the report object without a write —
`report.aggregation_dfs[...]` (used by `impulse-ml`) — which is what makes sinkless mode useful.

## Sinkless mode (compute without writing)

Omit `unity_sink` from the config. `determine_report()` still computes events, aggregations, and
dimensions and exposes them on the report; `persist_results()` becomes a no-op. Good for notebooks,
tests, and feeding ML (see `impulse-ml`). Reading a DataFrame straight out of the query engine without
any report at all is `impulse-analyze`.

## Incremental processing

Reuses prior results for unchanged definitions and reprocesses only containers that are new or updated
in silver. Turn it on via `incremental.enabled` in config (see `impulse-config`) or pass
`is_incremental=True` to `determine_report()`.

**Mode resolution** (in order):

1. No gold layer yet → full run (nothing to compare against). The first run of a report is always full.
2. `config.incremental` set → `config.incremental.enabled` wins.
3. Otherwise the `is_incremental` argument wins.
4. Neither set → full.

**On each run:** each event/aggregation is compared against its stored `definition_hash`. Unchanged
definitions reprocess only new/updated containers (persisted via Delta `MERGE` on natural keys);
changed or brand-new definitions reprocess all matching containers (replaced atomically via
`replaceWhere` on `visual_id`/`event_id`/`channel_id`). A single run can mix both per entity.

**Forcing a full recalculation.** To recompute specific *unchanged* entities over all containers — e.g.
to backfill after a persistence fix or corrected upstream data — list their names under
`full_recalculation` in config (see `impulse-config`). Those aggregations/events/calculated channels are
treated as changed for that run regardless of their hash; everything else stays incremental. A no-op in
full mode, and a name matching no registered entity fails fast.

**What counts as a definition change** — only the hashed attributes; renames, descriptions, and units
are cosmetic and do not trigger reprocessing:

| Type                | Hashed                                             |
|---------------------|----------------------------------------------------|
| `BasicEvent`        | `expr` string                                      |
| `ContainerEvent`    | `name`                                             |
| `Histogram`         | `base_expr`, `bins`, `event`                       |
| `Histogram2D`       | `x_expr`, `y_expr`, `x_bins`, `y_bins`, `event`    |
| `StatsAggregator`   | `input_expressions`, `statistics`, `event`, `channel_names`, custom stats (incl. cross-channel `channel_name`) |
| `CalculatedChannel` | `expr` string + `identity`                         |

**Container-update detection** unions two sets: new containers (silver rows absent from gold) and
updated containers (silver `silver_last_modified_column` newer than the gold `gold_last_modified_column`).
If either column is missing on its side, update detection is skipped and only new containers are picked
up.

## Scheduling

A report pipeline is plain Python — put the lifecycle in a notebook or `.py` task and schedule it as a
Databricks Workflow to keep the gold layer fresh. Incremental mode makes repeated runs cheap.