1661. Average Time of Process per Machine

Problem¶

Given an Activity table with one start and one end row per (machine_id, process_id), return each machine's average process duration rounded to 3 decimals.

Input:

# Activity
-----------+------------+---------------+----------
machine_id | process_id | activity_type | timestamp
-----------+------------+---------------+----------
0          | 0          | 'start'       | 0.712
0          | 0          | 'end'         | 1.520
0          | 1          | 'start'       | 3.140
0          | 1          | 'end'         | 4.120
1          | 0          | 'start'       | 0.550
1          | 0          | 'end'         | 1.550

Output:

-----------+----------------
machine_id | processing_time
-----------+----------------
0          | 0.894
1          | 1.000

Key trick¶

Convert each process into one duration row by summing signed timestamps:

end contributes +timestamp
start contributes -timestamp

So each (machine_id, process_id) duration is \(\text{duration} = \text{end} - \text{start}\).

Trap¶

Joining only on process_id and forgetting machine_id.
Averaging raw timestamps instead of durations.
Dividing by row count instead of process count.
In PostgreSQL, ROUND(double precision, integer) needs a cast to numeric.

Why is it interesting?¶

It tests whether you can transform paired event rows into per-entity metrics, a common pattern in logs, sessions, jobs, and lifecycle tables.

SQL solution¶

SQLite¶

WITH process_duration AS (
    SELECT
        machine_id,
        process_id,
        -- Each process has exactly one start and one end.
        -- start is negative, end is positive, so SUM gives end - start.
        SUM(
            CASE
                WHEN activity_type = 'end' THEN timestamp
                ELSE -timestamp
            END
        ) AS duration
    FROM Activity
    GROUP BY machine_id, process_id
)
SELECT
    machine_id,
    ROUND(AVG(duration), 3) AS processing_time
FROM process_duration
GROUP BY machine_id;

PostgreSQL¶

WITH process_duration AS (
    SELECT
        machine_id,
        process_id,
        -- Same logic as SQLite.
        SUM(
            CASE
                WHEN activity_type = 'end' THEN timestamp
                ELSE -timestamp
            END
        ) AS duration
    FROM Activity
    GROUP BY machine_id, process_id
)
SELECT
    machine_id,
    -- PostgreSQL needs numeric for ROUND(value, decimals).
    ROUND(AVG(duration)::numeric, 3) AS processing_time
FROM process_duration
GROUP BY machine_id;

Pandas solution¶

import pandas as pd


def get_average_time(activity: pd.DataFrame) -> pd.DataFrame:
    df = activity.copy()

    # Use signed timestamps so each grouped process sums to end - start.
    df["signed_timestamp"] = df["timestamp"].where(
        df["activity_type"].eq("end"),
        -df["timestamp"],
    )

    process_duration = (
        df.groupby(["machine_id", "process_id"], as_index=False)
        .agg(duration=("signed_timestamp", "sum"))
    )

    result = (
        process_duration.groupby("machine_id", as_index=False)
        .agg(processing_time=("duration", "mean"))
    )

    result["processing_time"] = result["processing_time"].round(3)

    return result[["machine_id", "processing_time"]]

Pytest test¶

import sqlite3

import pandas as pd
import pytest
from pandas.testing import assert_frame_equal


SQLITE_QUERY = """
WITH process_duration AS (
    SELECT
        machine_id,
        process_id,
        SUM(
            CASE
                WHEN activity_type = 'end' THEN timestamp
                ELSE -timestamp
            END
        ) AS duration
    FROM Activity
    GROUP BY machine_id, process_id
)
SELECT
    machine_id,
    ROUND(AVG(duration), 3) AS processing_time
FROM process_duration
GROUP BY machine_id
ORDER BY machine_id;
"""


def get_average_time(activity: pd.DataFrame) -> pd.DataFrame:
    df = activity.copy()

    df["signed_timestamp"] = df["timestamp"].where(
        df["activity_type"].eq("end"),
        -df["timestamp"],
    )

    process_duration = (
        df.groupby(["machine_id", "process_id"], as_index=False)
        .agg(duration=("signed_timestamp", "sum"))
    )

    result = (
        process_duration.groupby("machine_id", as_index=False)
        .agg(processing_time=("duration", "mean"))
    )

    result["processing_time"] = result["processing_time"].round(3)

    return result[["machine_id", "processing_time"]].sort_values("machine_id").reset_index(drop=True)


def run_sqlite(activity: pd.DataFrame) -> pd.DataFrame:
    with sqlite3.connect(":memory:") as conn:
        activity.to_sql("Activity", conn, index=False)
        return pd.read_sql_query(SQLITE_QUERY, conn)


@pytest.mark.parametrize(
    "rows, expected_rows",
    [
        (
            [
                [0, 0, "start", 0.712],
                [0, 0, "end", 1.520],
                [0, 1, "start", 3.140],
                [0, 1, "end", 4.120],
                [1, 0, "start", 0.550],
                [1, 0, "end", 1.550],
                [1, 1, "start", 0.430],
                [1, 1, "end", 1.420],
                [2, 0, "start", 4.100],
                [2, 0, "end", 4.512],
                [2, 1, "start", 2.500],
                [2, 1, "end", 5.000],
            ],
            [
                [0, 0.894],
                [1, 0.995],
                [2, 1.456],
            ],
        ),
        (
            [
                [7, 0, "start", 10.0],
                [7, 0, "end", 13.0],
            ],
            [
                [7, 3.000],
            ],
        ),
        (
            [
                [1, 10, "start", 0.0],
                [1, 10, "end", 2.0],
                [1, 11, "start", 5.0],
                [1, 11, "end", 8.0],
                [2, 10, "start", 1.0],
                [2, 10, "end", 5.0],
                [2, 11, "start", 10.0],
                [2, 11, "end", 12.0],
            ],
            [
                [1, 2.500],
                [2, 3.000],
            ],
        ),
    ],
)
def test_average_time_sqlite_and_pandas(rows, expected_rows):
    activity = pd.DataFrame(
        rows,
        columns=["machine_id", "process_id", "activity_type", "timestamp"],
    )

    expected = pd.DataFrame(
        expected_rows,
        columns=["machine_id", "processing_time"],
    )

    sql_result = run_sqlite(activity)
    pandas_result = get_average_time(activity)

    assert_frame_equal(sql_result, expected, check_dtype=False, atol=1e-9)
    assert_frame_equal(pandas_result, expected, check_dtype=False, atol=1e-9)

Comment on my solution¶

Your SQL solution is correct and readable.
It is PostgreSQL-specific because of ::numeric; SQLite would use ROUND(AVG(duration), 3).
The self-join approach is a good interview answer, though conditional aggregation is shorter.
Your Pandas solution is correct and clear.
The merge keys are correct because you join on both machine_id and process_id.

-- WORKS
WITH StartActivity AS (
    SELECT
        machine_id,
        process_id,
        timestamp AS started_at
    FROM Activity
    WHERE activity_type = 'start'
),
EndActivity AS (
    SELECT
        machine_id,
        process_id,
        timestamp AS ended_at
    FROM Activity
    WHERE activity_type = 'end'
),
StartEndActivity AS (
    SELECT
        s.machine_id,
        s.process_id,
        e.ended_at - s.started_at AS duration
    FROM StartActivity AS s
    JOIN EndActivity AS e
      ON e.machine_id = s.machine_id
     AND e.process_id = s.process_id
)
SELECT
    machine_id,
    ROUND(AVG(duration)::numeric, 3) AS processing_time
FROM StartEndActivity
GROUP BY machine_id;

import pandas as pd

def get_average_time(activity: pd.DataFrame) -> pd.DataFrame:
    start_activity = activity.loc[activity["activity_type"] == "start"]
    end_activity = activity.loc[activity["activity_type"] == "end"]
    result = start_activity.merge(
        end_activity,
        on=["machine_id", "process_id"],
        how="inner",
        suffixes=("", "_end")
    )
    result["duration"] = result["timestamp_end"] - result["timestamp"]
    result = (
        result.groupby("machine_id", as_index=False)
        .agg(processing_time=("duration", "mean"))
    )
    result["processing_time"] = result["processing_time"].round(3)
    return result[["machine_id", "processing_time"]]