Skip to content

Querying event data

The events your data collection sends are archived on S3 as Parquet, and the usual way to analyse them from a task or a notebook is DuckDB — it reads the Parquet directly from S3, so there is no cluster to spin up and no data to copy first.

pipe_algorithms_lib.duckdb_utils provides the three pieces you need:

Function Purpose
prepare_parquet_path(codops, sitekey, days, event_type) resolve the S3 paths for a lookback window into a read_parquet(...) expression
execute_query(sql) run a SQL string, get back a list[tuple]
get_duckdb_connection() the cached connection, if you ever need it directly

duckdb (and is-bot, if you use the precise bot filter below) has to be in your environment.

A first query

from pipe_algorithms_lib.duckdb_utils import execute_query, prepare_parquet_path

CODOPS = "your_codops"
SITEKEY = "your_sitekey"        # e.g. dedw000000000128

parquet = prepare_parquet_path(
    codops=CODOPS,
    sitekey=SITEKEY,
    days=1,                     # 0 = today only, 1 = today + yesterday
    event_type="media_play",    # or page_view, or "*" for everything
)

rows = execute_query(f"""
    SELECT event['props']['pageType'] AS model_type,
           COUNT(*) AS event_count
    FROM {parquet}
    GROUP BY model_type
""")

prepare_parquet_path returns a string, so it goes straight into the FROM clause of an f-string query.

Where the data actually sits

s3://peach-members-{codops}/events_by_type/{event_type}/
    daily/{sitekey}_events/{YYYY-MM-DD}/*.parquet     ← past days
    hourly/{sitekey}_events/{YYYY-MM-DD}/*.parquet    ← today

Today is written hourly and past days are consolidated daily; prepare_parquet_path picks the right granularity per day for you. With event_type="*" it reads the combined peach-events prefix instead of a per-type one.

The fields you will use

The Parquet columns are nested, and addressed with ['...']:

Expression What it is
event['metadata']['id'] content id
event['props']['pageType'] content type — ARTICLE, VIDEO, AUDIO, …
event['metadata']['appName'] source app, e.g. nest, native-mobile
event['props']['categories'][1] category id — arrays are 1-indexed
event['props']['regions'][1] region, formatted CONTINENT::COUNTRY (EUROPE::DE)
client['id'] device / user id
client['device']['type'] device type — what bot filtering looks at
collect_timestamp event time, in milliseconds

Two rules for any ranking query

Filter out bots

Bot traffic will otherwise dominate whatever you rank. A single ILIKE handles the overwhelming majority:

WHERE client['device']['type'] NOT ILIKE '%bot%'

When you need to be thorough, build a filter from the is_bot library's patterns instead — more accurate, and slower:

from is_bot._patterns import default_patterns

conditions = [
    f"NOT REGEXP_MATCHES(client['device']['type'], '{regex}', 'i')"
    for regex in default_patterns
    if "(?" not in regex          # DuckDB has no lookahead
]
precise_bot_filter = " AND ".join(conditions)

Count distinct users, not events

COUNT(DISTINCT client['id']) AS event_count   -- yes
COUNT(*)                     AS event_count   -- no: one replaying user inflates this

Raw event counts reward repeat plays by the same person, which is rarely the popularity you meant to measure.

Ranking patterns

Top N overall

import time

cutoff_ms = int(time.time() * 1000) - 24 * 60 * 60 * 1000

rows = execute_query(f"""
    SELECT event['metadata']['id'] AS content_id,
           event['props']['pageType'] AS model_type,
           COUNT(DISTINCT client['id']) AS event_count
    FROM {parquet}
    WHERE collect_timestamp >= {cutoff_ms}
      AND client['device']['type'] NOT ILIKE '%bot%'
    GROUP BY content_id, model_type
    ORDER BY event_count DESC
    LIMIT 10
""")

Top N per group

ROW_NUMBER() with PARTITION BY gives you a separate ranking per content type, category or region in one pass:

WITH events AS (
    SELECT event['metadata']['id'] AS content_id,
           event['props']['pageType'] AS model_type,
           COUNT(DISTINCT client['id']) AS event_count
    FROM {parquet}
    WHERE client['device']['type'] NOT ILIKE '%bot%'
    GROUP BY content_id, model_type
)
SELECT * FROM (
    SELECT *, ROW_NUMBER() OVER (
                 PARTITION BY model_type ORDER BY event_count DESC
              ) AS rank
    FROM events
) WHERE rank <= 20

Swap the PARTITION BY for category or region to slice it differently. For regions, SUBSTRING up to the :: gives you the continent on its own.

Turning rows into JSON

execute_query returns tuples, so name the columns once and zip them:

fields = ["content_id", "model_type", "event_count"]
data = [dict(zip(fields, row)) for row in rows]

That list is what you write to Redis for an endpoint to serve — see Build your first feature for the shape of the whole thing.