Pipelines in Python

1.Parse a CSV line with quotes

Write parse_csv_line(line) that splits one CSV line into a list of strings. Fields are separated by commas. A field wrapped in double quotes may contain commas, and inside it two double quotes ("") stand for one literal quote; the wrapping quotes are removed. Spaces are kept as they are. An empty line returns [""], and a trailing comma adds a final empty field.

2.Normalize timestamps to UTC

Write normalize_timestamps(values) for a list of strings. Accepted forms: 'YYYY-MM-DDTHH:MM:SS', the same with a space instead of T, or a bare 'YYYY-MM-DD' (midnight). The two time forms may end with 'Z' or an offset like '+05:30' or '-04:00'; no suffix means UTC. Convert each to UTC as 'YYYY-MM-DDTHH:MM:SSZ'. Use None for any string that is not a valid date/time in these forms.

3.Deduplicate keeping the latest

Write dedupe_latest(records, key, version) for a list of dicts. Keep one record per value of record[key]: the one with the largest record[version]. On a version tie, keep the record that comes later in the list. Return the kept records ordered by where each key first appears in the input.

4.Apply a change log

Write apply_cdc(table, ops). table maps a row id (string) to a row dict. Apply ops in list order; each op is {"op", "id"} plus "data" for insert and update. insert sets the row to a copy of data, replacing any existing row. update merges data's fields into an existing row and is ignored if the id is absent. delete removes the id if present. Return the new table; do not mutate the input.

5.Slowly changing dimension type 2

Write scd2(dim, updates, as_of). dim rows are {"key","value","valid_from","valid_to","current"}; updates are {"key","value"}; for a repeated key only the last counts. If the key's current row has that value, skip it. Else close that row if any (valid_to = as_of, current = False) and add a current row with valid_from = as_of, valid_to = None. Return old rows in order, then new rows sorted by key.

6.Incremental load with a watermark

Write incremental_load(rows, watermark). Each row has "id" and "updated_at", a string in the fixed form 'YYYY-MM-DDTHH:MM:SS' (so string order is time order). Select rows with updated_at strictly after watermark; a watermark of None selects all. Return {"rows": selected rows sorted by updated_at then id, "watermark": the largest selected updated_at, or the old watermark if nothing was selected}.

7.Validate rows against a schema

Write validate_rows(rows, schema). schema maps a column to "int", "float", "str" or "bool". A row is good if every schema column is present, not None, and of that type: bools never count as int or float, and an int is accepted as float. Extra columns are fine. Return {"good": good rows in input order, "bad": [{"row": index in rows, "errors": sorted failing column names}]}.

8.Partition records by date

Write partition_by_date(records, field). record[field] is a string starting with 'YYYY-MM-DD' (a time part may follow). Group records into buckets named 'dt=YYYY-MM-DD' by that date. Records whose field is missing, None or an empty string go to 'dt=unknown'. Return a dict from bucket name to its records in input order; only buckets with records appear.

9.Hash-partition keys

Write hash_partition(keys, n) for a list of string keys and n partitions. A key's hash is the sum of the character codes (ord) of its characters, and its partition is hash % n. Return a list of n lists where list i holds the keys of partition i in input order, duplicates included; empty partitions are empty lists.

10.Inner and left join

Write join_records(left, right, key, how) for two lists of dicts; how is "inner" or "left". Pair each left row with every right row with the same key value, outputting the left fields updated with the right row's fields (right wins on name clashes). A missing or None key never matches. With "left", an unmatched left row is output as a copy. Order: left order, then right order within each left row.

11.Sessionize events

Write sessionize(events, gap). Each event is {"user", "ts"} with ts in integer seconds; the list may be out of order. Per user, sort by ts; a new session starts when an event comes more than gap seconds after the previous one (exactly gap stays in the session). Return [{"user", "start", "end", "events"}] giving first ts, last ts and event count, sorted by user then start.

12.Running total per key

Write running_total(rows, key, order, value), like SUM(value) OVER (PARTITION BY key ORDER BY order). For each row add "running_total": the sum of value over rows with the same key up to and including this row in order (ties in order go by input position). A None value counts as 0. Return copies of the rows in the original input order.

13.Row number per group

Write row_number(rows, partition, order_by, descending). Within each group of equal row[partition], number rows 1, 2, ... by row[order_by], ascending, or descending if descending is True; equal order_by values keep their input order. Return copies of the rows with an added field "rn", sorted by partition value, then rn.

14.Previous value with lag

Write add_lag(rows, key, order, value), like LAG(value) OVER (PARTITION BY key ORDER BY order). Order values are unique within a key. Add "prev": the value of the previous row with the same key by order (even if that value is None), or None for the first; and "change": value minus prev, or None when either is None. Return copies of the rows in the original input order.

15.Flatten nested JSON

Write flatten_json(obj) for a nested dict. Each non-dict value becomes one column named by joining the keys on its path with dots, so {"a": {"b": 1}} gives {"a.b": 1}. Lists and None are values and are kept as they are, not expanded. A nested empty dict produces no column. Return the flat dict.

16.Unflatten dotted columns

Write unflatten(flat), the reverse of flattening: split each key on dots and nest the value, so {"a.b": 1, "a.c": 2} gives {"a": {"b": 1, "c": 2}}. Values are never dicts. If one key is also the parent path of another (like "a" and "a.b"), the input is inconsistent: return None. Return the nested dict, or {} for an empty input.

17.Merge small files in order

Write pack_files(sizes, target) to merge small files without reordering them. Walk the files in order, adding each file's index to the current batch while the batch total stays at most target; when the next file would push it over, close the batch and start a new one with that file. A file bigger than target sits alone. Return the list of batches (lists of indices); [] for no files.

18.Backfill missing days

Write fill_missing_days(rows, start, end, fill). rows are {"date": 'YYYY-MM-DD', "value"}. Return one {"date", "value"} per calendar day from start to end inclusive, in date order, using the row value for that day or fill if there is none. If a date repeats, the last row wins. Rows outside the range are ignored. Return [] if start is after end.

19.Detect late-arriving events

Write late_events(events, lag). events arrive in list order as {"id", "event_time"} (integer seconds). Before each event, the watermark is the largest event_time seen so far minus lag; there is no watermark before the first event. An event is late if its event_time is strictly below the current watermark. Every event updates the max seen. Return the late ids in arrival order.

20.Data freshness report

Write freshness_report(loaded, now, sla, default_sla). loaded maps a table to its last load time 'YYYY-MM-DDTHH:MM:SS' or None; now has the same format. lag_minutes is the whole minutes elapsed, rounded down. A table is stale if lag_minutes > its limit in sla, or default_sla if absent. A never-loaded table has lag None and is stale. Return [{"table","lag_minutes","stale"}] sorted by table.

21.Column null rates

Write null_rates(rows, columns). For each name in columns, compute the fraction of rows in which that column is missing, None, or the empty string "" (0 and False are real values). Round each rate to 4 decimals. Return a dict from column to rate; with no rows every rate is 0.0.

22.Upsert into a keyed table

Write merge_into(target, source, key), an upsert by version. Rows are dicts with key and an integer "version". Process source rows in order: if no row with that key exists yet, append a copy; if one exists, replace it with a copy of the source row only when the source version is strictly greater. Return the merged list: target rows in their positions, then inserted rows in insertion order.

23.DAG task levels

Write task_levels(deps) to plan parallel DAG runs. deps maps a task to the list of tasks it depends on; a task named only as a dependency has none. Level 0 holds tasks with no dependencies; a task's level is 1 + the highest level among its dependencies. Return a list of levels, each a sorted list of task names; [] for no tasks. If there is a cycle, return None.

24.Retry schedule with backoff

Write retry_schedule(base, factor, cap, max_retries, deadline). Retry i (counting from 0) waits base * factor**i seconds, but never more than cap. Plan at most max_retries retries, and stop before the first retry whose wait would make the total waiting time exceed deadline (reaching it exactly is allowed). Return the list of waits in order; [] if none fit.

25.Chunk an id range into batches

Write chunk_range(lo, hi, size, done) to split ids lo..hi (inclusive) into load batches. done is a list of [a, b] inclusive ranges already loaded (they may overlap or reach outside lo..hi); skip those ids. Cover each remaining run of consecutive ids with [start, end] chunks of at most size ids, starting at the run's first id. Return the chunks in ascending order; [] if nothing remains or lo > hi.

26.Compare two snapshots

Write diff_snapshots(old, new, key) for two snapshots, each a list of dicts with a unique value in field key. Return {"added": keys only in new, "removed": keys only in old, "changed": keys in both whose rows differ in any field, including a field present in only one of them}, each list sorted ascending.

27.Compact an event log

Write compact_log(events) to compact a keyed changelog. Each event is {"key", "seq", "value"}; events may be out of order and seq is unique per key. For each key keep the value of its highest-seq event. A None value is a tombstone: if a key's latest event is a tombstone, leave the key out. Return a dict from key to value.

28.Count events per time window

Write tumbling_counts(timestamps, width) for integer epoch seconds in any order. Each timestamp falls in the window starting at t - t % width, so a timestamp equal to a window start belongs to that window. Return [[window_start, count]] for every window from the earliest to the latest non-empty one, in order, with 0 for empty windows in between; [] for no timestamps.