Getting Started¶

This is the "Getting Started" streamdaq v2 notebook.

In [1]:

!uv pip install --quiet streamdaq
!uv pip show streamdaq

!uv pip install --quiet streamdaq !uv pip show streamdaq

error: No virtual environment found; run `uv venv` to create an environment, or
       pass `--system` to install into a non-virtual environment

Using Python 3.11.17 environment at: /opt/hostedtoolcache/Python/3.11.17/x64
Name: streamdaq
Version: 2.0.3
Location: /opt/hostedtoolcache/Python/3.11.17/x64/lib/python3.11/site-packages
Requires: datasketch, datasketches, dill, fastapi, lmdb, pathway, pydantic, strenum, uvicorn
Required-by:

In [2]:

# pip install streamdaq
import multiprocessing

import pathway as pw

from streamdaq.checks import InRange, WindowDataQualityCheck
from streamdaq.measures import Count, DistinctCount, MostFrequent
from streamdaq.sessions import Session
from streamdaq.tasks import Task


def main():
    # Step 1: Point Stream DaQ to your data stream (here, a tiny in-memory stream)
    def read_stream() -> pw.Table:
        return pw.debug.table_from_markdown(
            """
            user_id | timestamp | interaction_events
            UserA   | 1         | 7
            UserA   | 2         | 3
            UserB   | 3         | 7
            UserA   | 4         | 12
            UserB   | 5         | 5
            UserB   | 6         | 7
            UserA   | 7         | 9
            UserB   | 8         | 7
            UserA   | 11        | 2
            UserB   | 12        | 4
            UserA   | 13        | 4
            UserB   | 14        | 1
            """
        )

    def is_seven_frequent(most_frequent: tuple) -> bool:
        return 7 in most_frequent

    def print_to_terminal(table: pw.Table) -> None:
        pw.io.subscribe(table, on_change=lambda key, row, time, is_addition: print(row))

    # Step 2: Create a monitoring task that reads the stream and prints the results
    task = Task(
        name="interactions",
        input=read_stream,
        output=print_to_terminal,
        windowby_column="timestamp",
        include_window_bounds=True,
    )

    # Step 3: Define what Data Quality means for you
    # Per-row checks, evaluated on every incoming element
    task.add_instant_checks(
        InRange(name="valid_events", column="interaction_events", low=0, high=10),
    )
    # Per-window checks, evaluated on every tumbling window of 10 time units
    task.add_window_checks(
        WindowDataQualityCheck("interaction_count", Count("interaction_events"), "(5, 15]"),
        WindowDataQualityCheck("my_freq", MostFrequent("interaction_events"), is_seven_frequent),
        WindowDataQualityCheck("interaction_distinct", DistinctCount("interaction_events"), "<= 5"),
        window=pw.temporal.tumbling(duration=10),
    )

    # Step 4: Kick-off monitoring and let Stream DaQ do the work while you focus on the important
    Session(tasks=[task]).start()

    # Only needed in Jupyter notebooks: wait for the monitoring to finish, so its results
    # appear in the notebook. A plain Python script waits for it automatically on exit.
    for process in multiprocessing.active_children():
        process.join()


if __name__ == "__main__":
    main()

# pip install streamdaq import multiprocessing import pathway as pw from streamdaq.checks import InRange, WindowDataQualityCheck from streamdaq.measures import Count, DistinctCount, MostFrequent from streamdaq.sessions import Session from streamdaq.tasks import Task def main(): # Step 1: Point Stream DaQ to your data stream (here, a tiny in-memory stream) def read_stream() -> pw.Table: return pw.debug.table_from_markdown( """ user_id | timestamp | interaction_events UserA | 1 | 7 UserA | 2 | 3 UserB | 3 | 7 UserA | 4 | 12 UserB | 5 | 5 UserB | 6 | 7 UserA | 7 | 9 UserB | 8 | 7 UserA | 11 | 2 UserB | 12 | 4 UserA | 13 | 4 UserB | 14 | 1 """ ) def is_seven_frequent(most_frequent: tuple) -> bool: return 7 in most_frequent def print_to_terminal(table: pw.Table) -> None: pw.io.subscribe(table, on_change=lambda key, row, time, is_addition: print(row)) # Step 2: Create a monitoring task that reads the stream and prints the results task = Task( name="interactions", input=read_stream, output=print_to_terminal, windowby_column="timestamp", include_window_bounds=True, ) # Step 3: Define what Data Quality means for you # Per-row checks, evaluated on every incoming element task.add_instant_checks( InRange(name="valid_events", column="interaction_events", low=0, high=10), ) # Per-window checks, evaluated on every tumbling window of 10 time units task.add_window_checks( WindowDataQualityCheck("interaction_count", Count("interaction_events"), "(5, 15]"), WindowDataQualityCheck("my_freq", MostFrequent("interaction_events"), is_seven_frequent), WindowDataQualityCheck("interaction_distinct", DistinctCount("interaction_events"), "<= 5"), window=pw.temporal.tumbling(duration=10), ) # Step 4: Kick-off monitoring and let Stream DaQ do the work while you focus on the important Session(tasks=[task]).start() # Only needed in Jupyter notebooks: wait for the monitoring to finish, so its results # appear in the notebook. A plain Python script waits for it automatically on exit. for process in multiprocessing.active_children(): process.join() if __name__ == "__main__": main()

{'user_id': 'UserA', 'timestamp': 2, 'interaction_events': 3, 'valid_events': True, '_streamdaq_metadata': pw.Json({'task_name': 'interactions', 'task_version': 1})}

{'user_id': 'UserB', 'timestamp': 3, 'interaction_events': 7, 'valid_events': True, '_streamdaq_metadata': pw.Json({'task_name': 'interactions', 'task_version': 1})}

{'user_id': 'UserA', 'timestamp': 1, 'interaction_events': 7, 'valid_events': True, '_streamdaq_metadata': pw.Json({'task_name': 'interactions', 'task_version': 1})}

{'user_id': 'UserB', 'timestamp': 12, 'interaction_events': 4, 'valid_events': True, '_streamdaq_metadata': pw.Json({'task_name': 'interactions', 'task_version': 1})}

{'user_id': 'UserA', 'timestamp': 4, 'interaction_events': 12, 'valid_events': False, '_streamdaq_metadata': pw.Json({'task_name': 'interactions', 'task_version': 1})}

{'user_id': 'UserB', 'timestamp': 8, 'interaction_events': 7, 'valid_events': True, '_streamdaq_metadata': pw.Json({'task_name': 'interactions', 'task_version': 1})}

{'user_id': 'UserB', 'timestamp': 5, 'interaction_events': 5, 'valid_events': True, '_streamdaq_metadata': pw.Json({'task_name': 'interactions', 'task_version': 1})}

{'user_id': 'UserA', 'timestamp': 11, 'interaction_events': 2, 'valid_events': True, '_streamdaq_metadata': pw.Json({'task_name': 'interactions', 'task_version': 1})}

{'user_id': 'UserA', 'timestamp': 13, 'interaction_events': 4, 'valid_events': True, '_streamdaq_metadata': pw.Json({'task_name': 'interactions', 'task_version': 1})}

{'user_id': 'UserB', 'timestamp': 6, 'interaction_events': 7, 'valid_events': True, '_streamdaq_metadata': pw.Json({'task_name': 'interactions', 'task_version': 1})}

{'user_id': 'UserA', 'timestamp': 7, 'interaction_events': 9, 'valid_events': True, '_streamdaq_metadata': pw.Json({'task_name': 'interactions', 'task_version': 1})}

{'user_id': 'UserB', 'timestamp': 14, 'interaction_events': 1, 'valid_events': True, '_streamdaq_metadata': pw.Json({'task_name': 'interactions', 'task_version': 1})}

{'window_start': 0, 'window_end': 10, 'interaction_count': True, 'my_freq': True, 'interaction_distinct': True, '_streamdaq_metadata': pw.Json({'task_name': 'interactions', 'task_version': 1})}

{'window_start': 10, 'window_end': 20, 'interaction_count': False, 'my_freq': False, 'interaction_distinct': True, '_streamdaq_metadata': pw.Json({'task_name': 'interactions', 'task_version': 1})}