path = untar_data(URLs.ML_100k)data
Data loading, tokenization, and dataset assembly for fastpragma
DataSource
Declare each data source with its column types. Profile sources omit time_col — all fields share the eval-point timestamp. Event sources provide a time_col for ordering.
DataSource
def DataSource(
df:LazyFrame, cats:list=None, conts:list=None, signed_conts:list=None, texts:list=None, time_col:str=None,
lifelong:list=None, num_buckets:int=100, cardinality_threshold:int=100, entity_col:str='entity_id',
is_profile:bool=False, name:str=None
):Initialize self. See help(type(self)) for accurate signature.
# Test 1: Explicit column types
df = pl.LazyFrame({'cat_col': ['a', 'b', 'a'], 'num_col': [1.0, 2.0, 3.0], 'id_col': [10, 20, 30]})
s = DataSource(df, cats=['cat_col'], conts=['num_col'], entity_col='id_col')
test_eq(sorted(s.cats), ['cat_col'])
test_eq(s.conts, ['num_col'])
# Test 2: Explicit overrides take priority
s2 = DataSource(df, cats=['cat_col'], conts=['num_col', 'id_col'], entity_col='id_col')
test_eq(sorted(s2.cats), ['cat_col'])
test_eq(sorted(s2.conts), ['id_col', 'num_col'])
# Test 3: time_col is excluded from cats/conts
df3 = pl.LazyFrame({'id_col': [1,2], 'evt': ['x', 'y'], 'val': [1, 2], 'ts': [100, 200]})
s3 = DataSource(df3, cats=['evt'], conts=['val'], time_col='ts', entity_col='id_col')
test_eq('ts' not in s3.cats + s3.conts, True)
# Test 4: Float32
df4 = pl.LazyFrame({'f32': [1.0, 2.0]}, schema={'f32': pl.Float32})
s4 = DataSource(df4, conts=['f32'], entity_col='f32')
test_eq(s4.conts, ['f32'])profile = DataSource.from_file(path / "u.user", separator="|", has_header=False, new_columns=["user_id","age","gender","occupation","zip_code"],
cats=["gender","occupation"], conts=["age"], entity_col="user_id")
profileDataSource(columns=['user_id', 'age', 'gender', 'occupation', 'zip_code'], name=u cats=['gender', 'occupation'], conts=['age'], texts=[], time_col=None)
shape: (5, 5)
┌─────────┬─────┬────────┬────────────┬──────────┐
│ user_id ┆ age ┆ gender ┆ occupation ┆ zip_code │
│ --- ┆ --- ┆ --- ┆ --- ┆ --- │
│ i64 ┆ i64 ┆ str ┆ str ┆ str │
╞═════════╪═════╪════════╪════════════╪══════════╡
│ 1 ┆ 24 ┆ M ┆ technician ┆ 85711 │
│ 2 ┆ 53 ┆ F ┆ other ┆ 94043 │
│ 3 ┆ 23 ┆ M ┆ writer ┆ 32067 │
│ 4 ┆ 24 ┆ M ┆ technician ┆ 43537 │
│ 5 ┆ 33 ┆ F ┆ other ┆ 15213 │
└─────────┴─────┴────────┴────────────┴──────────┘pdf = pd.DataFrame({'cat': ['x','y','x'], 'val': [1,2,3], 'id': [10,20,30]})
s5 = DataSource.from_df(pdf, cats=['cat'], conts=['val'], entity_col='id')
test_eq(sorted(s5.cats), ['cat'])
test_eq(s5.conts, ['val'])
test_eq(s5.entity_col, 'id')
test_eq(s5.time_col, None)Key-Value-Time Conversion
Convert wide DataFrames to the uniform (key, value, value_type, time) format used internally. Profile rows get time=0 (all at eval point); events use their timestamp column; lifelong columns carry their own datetime.
Tokenizer
def Tokenizer(
num_buckets:int=100, cardinality_threshold:int=100
):Initialize self. See help(type(self)) for accurate signature.
Tokenizer
Builds key and value vocabularies, bucketizes numerical values, and tokenizes source schemas into fixed-size tensors for the model.
- Each key gets one token in the key vocab
- Categorical values: one token per unique value
- Numerical values: percentile-bucket index
- Textual values: BPE subword tokens
Keys use a single shared vocabulary across all sources. replace_strict maps known keys to their index; any unknown key gets the [UNK] fallback.
Categorical values map one-to-one to vocabulary tokens. PRAGMA uses this for high-frequency fields like MCC codes — one value, one token, direct lookup.
Continuous values are discretized into percentile bins learned per key during fit().
0maps to the special numerical bucket<bucket_0>- null / missing / unparsable values map to
[UNK] - non-zero numeric values map to percentile buckets
<bucket_1>,<bucket_2>, … np.searchsortedfinds the bucket index from the precomputed boundaries
This keeps all model inputs discrete while preserving approximate magnitude/order information.
tok = Tokenizer(num_buckets=4)
for i in range(tok.num_buckets+1): tok._add_val(f'<bucket_{i}>')
tok.num_bounds['amt'] = np.array([2.5, 5.0, 7.5])
kvt = pl.LazyFrame(dict(
key=['amt','amt','amt','amt','amt','amt','val'],
value=[None,'0.0','2.5','5.0','-3.0','999.0','9'],
vtype=['num']*7,
time=[0]*7))
result = tok._map_num_vals(kvt).collect()
want = {
None: '[UNK]',
'0.0': '<bucket_0>',
'2.5': '<bucket_1>',
'5.0': '<bucket_2>',
'-3.0': '<bucket_1>',
'999.0': '<bucket_4>',
}
for v,b in want.items(): test_eq(result.filter(pl.col('key')=='amt', pl.col('value').eq_missing(v))['val_id'].item(), tok.val_vocab[b])
test_eq(result.filter(pl.col('key')=='val')['val_id'].item(), tok.val_vocab['[UNK]'])result
shape: (7, 6)
| key | value | vtype | time | val_id | val_pos |
|---|---|---|---|---|---|
| str | str | str | i64 | i64 | i64 |
| "amt" | null | "num" | 0 | 2 | 0 |
| "amt" | "0.0" | "num" | 0 | 3 | 0 |
| "amt" | "2.5" | "num" | 0 | 4 | 0 |
| "amt" | "5.0" | "num" | 0 | 5 | 0 |
| "amt" | "-3.0" | "num" | 0 | 4 | 0 |
| "amt" | "999.0" | "num" | 0 | 7 | 0 |
| "val" | "9" | "num" | 0 | 2 | 0 |
High-cardinality text gets BPE subword tokenization via tiktoken. Each subword becomes its own row sharing the same key — the model groups them later via shared key embeddings and within-field positional encodings.
Using a dynamic bpe_offset = len(val_vocab) computed at encode time means BPE IDs naturally shift as val_vocab grows — add a new categorical token, and all subword IDs move up by one. That’s fine as long as you re-encode the entire dataset whenever you refit. Unknown keys/values/new numeric fields all route to [UNK] gracefully, so a new source won’t error — it’ll just need refitting (and re-encoding) to learn from it properly.
tok = Tokenizer()
tok._add_key('desc'); tok._add_key('desc_short'); tok._add_val('hello')
tok.lowcard_text_keys.add('desc_short')
class MockEncoded:
def __init__(self, ids): self.ids = ids
class MockTok:
def encode(self, text): return MockEncoded([7,8,7] if 'plan' in text else [9])
tok.bpe_tokenizer = MockTok()
off = 0
kvt = pl.LazyFrame(dict(
key=['desc', 'desc', 'desc_short'],
value=['metal plan', 'world', 'hello'],
vtype=['text']*3,
time=[0]*3))
res = tok._map_text_vals(kvt).collect()
mp = res.filter(pl.col('key')=='desc', pl.col('value')=='metal plan')
w = res.filter(pl.col('key')=='desc', pl.col('value')=='world')
ds = res.filter(pl.col('key')=='desc_short')
test_eq(mp['val_id'].to_list(), [off+7, off+8, off+7])
test_eq(mp['val_pos'].to_list(), [0,1,2])
test_eq(w['val_id'].item(), off+9)
test_eq(w['val_pos'].item(), 0)
test_eq(ds['val_id'].item(), tok.val_vocab['hello'])
test_eq(ds['val_pos'].item(), 0)
test_eq(res.height, 5)result
shape: (7, 6)
| key | value | vtype | time | val_id | val_pos |
|---|---|---|---|---|---|
| str | str | str | i64 | i64 | i64 |
| "amt" | null | "num" | 0 | 2 | 0 |
| "amt" | "0.0" | "num" | 0 | 3 | 0 |
| "amt" | "2.5" | "num" | 0 | 4 | 0 |
| "amt" | "5.0" | "num" | 0 | 5 | 0 |
| "amt" | "-3.0" | "num" | 0 | 4 | 0 |
| "amt" | "999.0" | "num" | 0 | 7 | 0 |
| "val" | "9" | "num" | 0 | 2 | 0 |
A single-pass scan builds all vocabularies and parameters: key vocab across all sources, val vocab from categoricals, percentile boundaries for continuous columns, and identifies low-cardinality text fields to treat as categorical.
Time Features
Tokenization keeps event timestamps as raw epoch seconds until this stage. Events use log-seconds to the latest event in the entity history; lifelong profile fields use log-seconds to the evaluation point; regular profile fields stay at zero.
Truncation (max events per user, max tokens per event) belongs in the dataloader, not in tokenization. Tokenization produces the full flat parquet per source; the dataloader applies truncation at load time so it can be tuned per experiment without re-tokenizing.
profile = DataSource.from_file(path / "u.user", separator="|", has_header=False,
new_columns=["user_id","age","gender","occupation","zip_code"],
cats=["gender","occupation"], conts=["age"], entity_col="user_id",
is_profile=True)
events = DataSource.from_file(path / "u.data", separator="\t", has_header=False,
new_columns=["user_id", "movie_id", "rating", "timestamp"],
cats=["movie_id"], conts=["rating"], time_col="timestamp", entity_col="user_id")tok = Tokenizer(num_buckets=10, cardinality_threshold=100)
tok.fit(L([profile, events]))
profile_tok = tok._tokenize(profile, eval_time='1998-04-01T00:00:00').head(3).collect()
events_tok = tok._tokenize(events, eval_time='1998-04-01T00:00:00').head(2).collect()
print("Profile:", profile_tok)
print("Events:", events_tok)Profile: shape: (3, 9)
┌─────────┬────────┬────────┬───────┬───┬─────────┬───────┬──────┬────────┐
│ user_id ┆ key ┆ key_id ┆ value ┆ … ┆ val_pos ┆ vtype ┆ time ┆ logsec │
│ --- ┆ --- ┆ --- ┆ --- ┆ ┆ --- ┆ --- ┆ --- ┆ --- │
│ i64 ┆ str ┆ i64 ┆ str ┆ ┆ i64 ┆ str ┆ i64 ┆ f64 │
╞═════════╪════════╪════════╪═══════╪═══╪═════════╪═══════╪══════╪════════╡
│ 1 ┆ gender ┆ 6 ┆ M ┆ … ┆ 0 ┆ cat ┆ 0 ┆ 0.0 │
│ 2 ┆ gender ┆ 6 ┆ F ┆ … ┆ 0 ┆ cat ┆ 0 ┆ 0.0 │
│ 3 ┆ gender ┆ 6 ┆ M ┆ … ┆ 0 ┆ cat ┆ 0 ┆ 0.0 │
└─────────┴────────┴────────┴───────┴───┴─────────┴───────┴──────┴────────┘
Events: shape: (2, 13)
┌─────────┬──────────┬────────┬───────┬───┬────────────┬──────┬─────┬──────┐
│ user_id ┆ key ┆ key_id ┆ value ┆ … ┆ logsec ┆ hour ┆ dow ┆ dom │
│ --- ┆ --- ┆ --- ┆ --- ┆ ┆ --- ┆ --- ┆ --- ┆ --- │
│ i64 ┆ str ┆ i64 ┆ str ┆ ┆ f64 ┆ f32 ┆ f32 ┆ f32 │
╞═════════╪══════════╪════════╪═══════╪═══╪════════════╪══════╪═════╪══════╡
│ 1 ┆ movie_id ┆ 7 ┆ 168 ┆ … ┆ 115.438142 ┆ 21.0 ┆ 1.0 ┆ 22.0 │
│ 1 ┆ movie_id ┆ 7 ┆ 172 ┆ … ┆ 115.438142 ┆ 21.0 ┆ 1.0 ┆ 22.0 │
└─────────┴──────────┴────────┴───────┴───┴────────────┴──────┴─────┴──────┘Tokenizer Persistence
The fitted tokenizer is part of the dataset contract. Save key/value vocabularies, numeric bucket boundaries, low-cardinality text choices, and any BPE model so tokenized train/valid/test data can be reproduced exactly.
Tokenizer.save
def save(
path:str='tokenizer.json'
):Call self as a function.
Tokenizer.load
def load(
path:str='tokenizer.json'
):Call self as a function.
tmp = Path(tempfile.mkdtemp())/'tokenizer.json'
tok.save(tmp)
tok2 = Tokenizer.load(tmp)
test_eq(tok2.key_vocab, tok.key_vocab)
test_eq(tok2.val_vocab, tok.val_vocab)
test_eq(set(tok2.num_bounds), set(tok.num_bounds))
test_eq(tok2.bpe_offset, len(tok2.val_vocab))PRAGMADataset
Assemble profile + event sources into per-entity tokenized sequences. Fits a tokenizer across all sources and produces tokenized tensors keyed by entity.
Dataset Orchestration
PRAGMADataset owns source consistency, tokenizer fitting, and sharded parquet writing. It preserves the caller’s entity_col name and keeps truncation out of tokenization so dataloader experiments can change context limits without re-tokenizing.
PRAGMADataset
def PRAGMADataset(
profile:NoneType=None, events:NoneType=None, entity_col:str='entity_id', out_path:NoneType=None
):Initialize self. See help(type(self)) for accurate signature.
write_kv is the main function which does everything
write_kv is the final orchestration step that takes fitted DataSources, tokenizes them, combines them, assigns each entity to a shard, and writes parquet shards to disk.
The flow is:
- Ensure tokenizer exists
if self.tokenizer is None: self.fit_tokenizer()If the dataset does not already have a fitted tokenizer, it calls fit_tokenizer(). That builds key/value vocabs, numeric bucket boundaries, low-card text handling, and possibly BPE state across all sources.
- Compute global latest event timestamp per entity
entity_max_ts = _compute_entity_max_ts(self.events)This matters when there are multiple event sources. Each entity needs one shared max_epoch, otherwise each event source would incorrectly treat its own latest event as logsec=0.
- Create list of tokenized source parts
parts = []Everything tokenized gets appended here before concatenation.
- Tokenize profile source, if present
parts.append(self.tokenizer._tokenize(self.profile, eval_time=eval_time).with_columns(pl.lit(-1).alias('source_idx'), pl.lit(-1).alias('event_idx')))Profile rows have no event order, so they get:
source_idx = -1event_idx = -1
Inside _tokenize, profile fields get logsec=0 for normal profile fields, while lifelong fields get logsec based on distance from eval_time.
- Tokenize each event source
for i,src in enumerate(progress_bar(self.events)):
parts.append(self.tokenizer._tokenize(src, eval_time=eval_time, entity_max_ts=entity_max_ts).with_columns(pl.lit(i).alias('source_idx')))Each event source is tokenized with the shared entity_max_ts.
Inside _tokenize, event rows get:
logsec: time delta from entity’s global latest eventhourdowdom- existing
event_idx - key/value ids from the tokenizer
Each source also gets its own source_idx.
- Concatenate all tokenized sources
lf = pl.concat(parts, how='diagonal_relaxed')diagonal_relaxed allows profile and event sources to have different columns. Missing columns become null.
- Assign entity shard
pl.col(self.entity_col).hash().mod(n_shards).alias('shard_id')Each entity is deterministically assigned to a shard by hashing its entity id.
- Fill event time feature nulls
pl.col('hour').cast(pl.Float32).fill_null(0)
pl.col('dow').cast(pl.Float32).fill_null(0)
pl.col('dom').cast(pl.Float32).fill_null(0)Profile rows do not naturally have these fields, so they are filled with 0.
- Sort the combined lazy frame
.sort(self.entity_col, 'source_idx', 'event_idx', 'key', 'val_pos')This gives a stable order:
- group by entity
- profile first because
source_idx=-1 - then event sources by source index
- then events by event index
- then fields by key
- then text subword positions by
val_pos
- Write one parquet file per shard
for s in progress_bar(range(n_shards)):
lf.filter(pl.col('shard_id') == s).sink_parquet(self.out_path/f'shard_{s}.parquet')Each shard is written independently as:
shard_0.parquet
shard_1.parquet
...The important thing is that this stays memory-light because it uses lazy Polars and writes each filtered shard directly with sink_parquet.
- Return output path
return self.out_pathSo the overall responsibility of write_kv is:
Fit/load tokenizer if needed → tokenize profile/events → add time/source/shard metadata → sort → write sharded parquet files.
PRAGMADataset.write_kv
def write_kv(
eval_time, n_shards:int=100
):Tokenize sources and write to entity-sharded parquet files.
tmp = Path(tempfile.mkdtemp())
ds = PRAGMADataset(profile=psrc, events=[esrc], entity_col='uid', out_path=tmp)
tok = ds.fit_tokenizer(num_buckets=4)
test_eq(tok is ds.tokenizer, True)
test_eq((tmp/'tokenizer.json').exists(), True)
test_eq('plan' in tok.key_vocab, True)
test_eq('evt' in tok.key_vocab, True)
test_eq('amt' in tok.key_vocab, True)Keys: 9, Vals: 13, BPE: noneout = ds.write_kv(eval_time=datetime(2020,1,11,tzinfo=timezone.utc), n_shards=2)
test_eq(out, tmp)
test_eq(sorted(o.name for o in tmp.glob('*.parquet')), ['shard_0.parquet', 'shard_1.parquet'])tokenizing profile
█
|----------------------------------------| 0.00% [0/1 00:00<?]tokenizing event source 0: None
|████████████████████████████████████████| 100.00% [1/1 00:00<00:00]combining sources
█
|----------------------------------------| 0.00% [0/2 00:00<?]
|████████████████████--------------------| 50.00% [1/2 00:00<00:00]
|████████████████████████████████████████| 100.00% [2/2 00:00<00:00]PRAGMADataset.show_summary
def show_summary():Call self as a function.
out = Path(tempfile.mkdtemp())/'ml100k_tok'
ds = PRAGMADataset(profile=profile, events=[events], entity_col='user_id', out_path=out)
ds.show_summary()
shape: (2, 14)
| i | name | kind | rows | entities | entity_col | time_col | cats | conts | signed_conts | texts | lifelong | min_time | max_time |
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| i64 | str | str | i64 | i64 | str | str | i64 | i64 | i64 | i64 | i64 | i64 | i64 |
| 0 | "u" | "profile" | 943 | 943 | "user_id" | null | 2 | 1 | 0 | 0 | 0 | null | null |
| 1 | "u" | "event" | 100000 | 943 | "user_id" | "timestamp" | 1 | 1 | 0 | 0 | 0 | 874724710 | 893286638 |
from nbdev.export import nb_exportnb_export('01_data.ipynb')End-to-End Tokenization Check
MovieLens 100k is used as a compact integration test: fit the shared tokenizer, write sharded parquet, then verify row counts, token ids, profile/event conventions, and time features.
out = Path(tempfile.mkdtemp())/'ml100k_tok'
ds = PRAGMADataset(profile=profile, events=[events], entity_col='user_id', out_path=out)
print(ds.show_summary())
# tok = ds.fit_tokenizer(num_buckets=10, cardinality_threshold=100)
ds.write_kv(eval_time='1998-04-01T00:00:00', n_shards=4)
sorted(o.name for o in out.iterdir())shape: (2, 14)
┌─────┬──────┬─────────┬────────┬───┬───────┬──────────┬───────────┬───────────┐
│ i ┆ name ┆ kind ┆ rows ┆ … ┆ texts ┆ lifelong ┆ min_time ┆ max_time │
│ --- ┆ --- ┆ --- ┆ --- ┆ ┆ --- ┆ --- ┆ --- ┆ --- │
│ i64 ┆ str ┆ str ┆ i64 ┆ ┆ i64 ┆ i64 ┆ i64 ┆ i64 │
╞═════╪══════╪═════════╪════════╪═══╪═══════╪══════════╪═══════════╪═══════════╡
│ 0 ┆ u ┆ profile ┆ 943 ┆ … ┆ 0 ┆ 0 ┆ null ┆ null │
│ 1 ┆ u ┆ event ┆ 100000 ┆ … ┆ 0 ┆ 0 ┆ 874724710 ┆ 893286638 │
└─────┴──────┴─────────┴────────┴───┴───────┴──────────┴───────────┴───────────┘
Keys: 10, Vals: 1809, BPE: none
tokenizing profile
█
|----------------------------------------| 0.00% [0/1 00:00<?]tokenizing event source 0: u
|████████████████████████████████████████| 100.00% [1/1 00:00<00:00]combining sources
█
|----------------------------------------| 0.00% [0/4 00:00<?]
|██████████------------------------------| 25.00% [1/4 00:00<00:01]
|████████████████████--------------------| 50.00% [2/4 00:00<00:00]
|██████████████████████████████----------| 75.00% [3/4 00:01<00:00]
|████████████████████████████████████████| 100.00% [4/4 00:01<00:00]['shard_0.parquet',
'shard_1.parquet',
'shard_2.parquet',
'shard_3.parquet',
'tokenizer.json']tok = ds.fit_tokenizer(num_buckets=10, cardinality_threshold=100)tokenizer foundlen(tok.key_vocab), len(tok.val_vocab), tok.num_bounds.keys(), tok.bpe_tokenizer(10, 1809, dict_keys(['age', 'rating']), None)toks = pl.concat([pl.scan_parquet(out/f'shard_{i}.parquet') for i in range(4)]).collect()
prof = toks.filter(pl.col('source_idx') == -1)
evts = toks.filter(pl.col('source_idx') == 0)
prof.shape, evts.shape, prof.columns, evts.columns((2829, 15),
(200000, 15),
['user_id',
'key',
'key_id',
'value',
'val_id',
'val_pos',
'vtype',
'time',
'logsec',
'source_idx',
'event_idx',
'hour',
'dow',
'dom',
'shard_id'],
['user_id',
'key',
'key_id',
'value',
'val_id',
'val_pos',
'vtype',
'time',
'logsec',
'source_idx',
'event_idx',
'hour',
'dow',
'dom',
'shard_id'])test_eq(prof.height, 943*3)
test_eq(evts.height, 100000*2)
test_eq(prof['source_idx'].unique().to_list(), [-1])
test_eq(prof['event_idx'].unique().to_list(), [-1])
test_eq(evts['source_idx'].unique().to_list(), [0])
test_eq(set(prof['key'].unique().to_list()), {'gender','occupation','age'})
test_eq(set(evts['key'].unique().to_list()), {'movie_id','rating'})
test_eq((prof['val_id'] != tok.val_vocab['[UNK]']).all(), True)
test_eq((evts['val_id'] != tok.val_vocab['[UNK]']).all(), True)u = 1
prof.filter(pl.col('user_id') == u).sort('key')
shape: (3, 15)
| user_id | key | key_id | value | val_id | val_pos | vtype | time | logsec | source_idx | event_idx | hour | dow | dom | shard_id |
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| i64 | str | i64 | str | i64 | i64 | str | i64 | f64 | i32 | i64 | f32 | f32 | f32 | u64 |
| 1 | "age" | 5 | "24" | 1730 | 0 | "num" | 0 | 0.0 | -1 | -1 | 0.0 | 0.0 | 0.0 | 0 |
| 1 | "gender" | 6 | "M" | 1686 | 0 | "cat" | 0 | 0.0 | -1 | -1 | 0.0 | 0.0 | 0.0 | 0 |
| 1 | "occupation" | 8 | "technician" | 1706 | 0 | "cat" | 0 | 0.0 | -1 | -1 | 0.0 | 0.0 | 0.0 | 0 |
evts.filter(pl.col('user_id') == u).sort('event_idx','key','val_pos').head(12)
shape: (12, 15)
| user_id | key | key_id | value | val_id | val_pos | vtype | time | logsec | source_idx | event_idx | hour | dow | dom | shard_id |
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| i64 | str | i64 | str | i64 | i64 | str | i64 | f64 | i32 | i64 | f32 | f32 | f32 | u64 |
| 1 | "movie_id" | 7 | "168" | 759 | 0 | "cat" | 874965478 | 115.438142 | 0 | 0 | 21.0 | 1.0 | 22.0 | 0 |
| 1 | "rating" | 9 | "5" | 1787 | 0 | "num" | 874965478 | 115.438142 | 0 | 0 | 21.0 | 1.0 | 22.0 | 0 |
| 1 | "movie_id" | 7 | "172" | 767 | 0 | "cat" | 874965478 | 115.438142 | 0 | 1 | 21.0 | 1.0 | 22.0 | 0 |
| 1 | "rating" | 9 | "5" | 1787 | 0 | "num" | 874965478 | 115.438142 | 0 | 1 | 21.0 | 1.0 | 22.0 | 0 |
| 1 | "movie_id" | 7 | "165" | 726 | 0 | "cat" | 874965518 | 115.438121 | 0 | 2 | 21.0 | 1.0 | 22.0 | 0 |
| … | … | … | … | … | … | … | … | … | … | … | … | … | … | … |
| 1 | "rating" | 9 | "4" | 1753 | 0 | "num" | 874965556 | 115.4381 | 0 | 3 | 21.0 | 1.0 | 22.0 | 0 |
| 1 | "movie_id" | 7 | "196" | 793 | 0 | "cat" | 874965677 | 115.438035 | 0 | 4 | 22.0 | 1.0 | 22.0 | 0 |
| 1 | "rating" | 9 | "5" | 1787 | 0 | "num" | 874965677 | 115.438035 | 0 | 4 | 22.0 | 1.0 | 22.0 | 0 |
| 1 | "movie_id" | 7 | "166" | 737 | 0 | "cat" | 874965677 | 115.438035 | 0 | 5 | 22.0 | 1.0 | 22.0 | 0 |
| 1 | "rating" | 9 | "5" | 1787 | 0 | "num" | 874965677 | 115.438035 | 0 | 5 | 22.0 | 1.0 | 22.0 | 0 |
chk = evts.filter(pl.col('user_id') == 1, pl.col('key') == 'movie_id').sort('time')
chk.select('user_id','event_idx','time','logsec','hour','dow','dom').head(), chk.select('user_id','event_idx','time','logsec','hour','dow','dom').tail()(shape: (5, 7)
┌─────────┬───────────┬───────────┬────────────┬──────┬─────┬──────┐
│ user_id ┆ event_idx ┆ time ┆ logsec ┆ hour ┆ dow ┆ dom │
│ --- ┆ --- ┆ --- ┆ --- ┆ --- ┆ --- ┆ --- │
│ i64 ┆ i64 ┆ i64 ┆ f64 ┆ f32 ┆ f32 ┆ f32 │
╞═════════╪═══════════╪═══════════╪════════════╪══════╪═════╪══════╡
│ 1 ┆ 0 ┆ 874965478 ┆ 115.438142 ┆ 21.0 ┆ 1.0 ┆ 22.0 │
│ 1 ┆ 1 ┆ 874965478 ┆ 115.438142 ┆ 21.0 ┆ 1.0 ┆ 22.0 │
│ 1 ┆ 2 ┆ 874965518 ┆ 115.438121 ┆ 21.0 ┆ 1.0 ┆ 22.0 │
│ 1 ┆ 3 ┆ 874965556 ┆ 115.4381 ┆ 21.0 ┆ 1.0 ┆ 22.0 │
│ 1 ┆ 4 ┆ 874965677 ┆ 115.438035 ┆ 22.0 ┆ 1.0 ┆ 22.0 │
└─────────┴───────────┴───────────┴────────────┴──────┴─────┴──────┘,
shape: (5, 7)
┌─────────┬───────────┬───────────┬───────────┬──────┬─────┬──────┐
│ user_id ┆ event_idx ┆ time ┆ logsec ┆ hour ┆ dow ┆ dom │
│ --- ┆ --- ┆ --- ┆ --- ┆ --- ┆ --- ┆ --- │
│ i64 ┆ i64 ┆ i64 ┆ f64 ┆ f32 ┆ f32 ┆ f32 │
╞═════════╪═══════════╪═══════════╪═══════════╪══════╪═════╪══════╡
│ 1 ┆ 267 ┆ 889751711 ┆ 11.336528 ┆ 1.0 ┆ 5.0 ┆ 13.0 │
│ 1 ┆ 268 ┆ 889751712 ┆ 11.090355 ┆ 1.0 ┆ 5.0 ┆ 13.0 │
│ 1 ┆ 269 ┆ 889751712 ┆ 11.090355 ┆ 1.0 ┆ 5.0 ┆ 13.0 │
│ 1 ┆ 270 ┆ 889751736 ┆ 0.0 ┆ 1.0 ┆ 5.0 ┆ 13.0 │
│ 1 ┆ 271 ┆ 889751736 ┆ 0.0 ┆ 1.0 ┆ 5.0 ┆ 13.0 │
└─────────┴───────────┴───────────┴───────────┴──────┴─────┴──────┘)chk = evts.filter(pl.col('user_id') == 1, pl.col('key') == 'movie_id')
test_eq(chk.filter(pl.col('time') == pl.col('time').max())['logsec'].unique().item(), 0.0)
test_eq((chk['logsec'] >= 0).all(), True)
test_eq(chk.sort('time')['logsec'].to_list()[0] >= chk.sort('time')['logsec'].to_list()[-1], True)