November 1, 2022TUTORIAL · DATA-PIPELINE

Time between events in a multi-topic event stream

We are processing a stream of events on different topics. We want to compute, for each event, how much time has passed since the previous event on the same topic. The issue is that the events do not appear in the order of their timestamps at input! With Pathway there is no need to worry about that!

First we use Debezium to synchronize the input data from a database table with the following columns:

  • timestamp - logical date-time when the event happened
  • topic_id - topic in question
  • message - message content.
import pathway as pw

events = pw.io.debezium.read(
    rdkafka_settings={
        "group.id": "$GROUP_NAME",
        "bootstrap.servers": "clean-panther-8776-eu1-kafka.upstash.io:9092",
        "session.timeout.ms": "6000",
    },
    topics=["important_events"],
)

Then we sort the events from the table. Pathway provides a sort function to sort a table according to its key column: in this case we are going to sort according to the timestamps of the events. In addition, each topic is mapped to an instance field, which allows us to work on different streams simultaneously. The prev and next pointers are automatically extracted.

sorted_events = events.sort(key=events.timestamp, instance=events.topic_id)

Finally, we process events in order of their timestamps at input.

events_with_prev = events.having(sorted_events.prev)
differences = events_with_prev.select(
    delta=events.restrict(events_with_prev).timestamp - pw.this.timestamp
)

pw.io.postgres.write(
    differences,
    postgres_settings={
        "host": "localhost",
        "port": "5432",
        "dbname": "transactions",
        "user": "pathway",
        "password": "my_password",
    },
    table_name="events_processed",
)

Przemek Uznański

Data Structures and Optimization Expert

event streammulti-topicDebeziumorderingsort
Share new articles with me each month
Share this article