# data
``` python
path = untar_data(URLs.ML_100k)
```
## 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.
------------------------------------------------------------------------
source
### DataSource
``` python
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.*
``` python
# 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'])
```
``` python
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")
profile
```
DataSource(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 │
└─────────┴─────┴────────┴────────────┴──────────┘
``` python
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.
------------------------------------------------------------------------
source
### Tokenizer
``` python
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()`.
- `0` maps to the special numerical bucket ``
- null / missing / unparsable values map to `[UNK]`
- non-zero numeric values map to percentile buckets ``,
``, …
- `np.searchsorted` finds the bucket index from the precomputed
boundaries
This keeps all model inputs discrete while preserving approximate
magnitude/order information.
``` python
tok = Tokenizer(num_buckets=4)
for i in range(tok.num_buckets+1): tok._add_val(f'')
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': '',
'2.5': '',
'5.0': '',
'-3.0': '',
'999.0': '',
}
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]'])
```
``` python
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.
``` python
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)
```
``` python
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.
``` python
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")
```
``` python
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.
------------------------------------------------------------------------
source
### Tokenizer.save
``` python
def save(
path:str='tokenizer.json'
):
```
*Call self as a function.*
------------------------------------------------------------------------
source
### Tokenizer.load
``` python
def load(
path:str='tokenizer.json'
):
```
*Call self as a function.*
``` python
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.
## Shared Event Time Reference
When multiple event sources exist, each entity needs one global latest
timestamp across all event sources. Otherwise every source would
incorrectly treat its own latest event as `logsec=0`.
## Dataset Orchestration
[`PRAGMADataset`](https://risheekkumarb.github.io/fastpragma/data.html#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.
------------------------------------------------------------------------
source
### PRAGMADataset
``` python
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
[`DataSource`](https://risheekkumarb.github.io/fastpragma/data.html#datasource)s,
tokenizes them, combines them, assigns each entity to a shard, and
writes parquet shards to disk.
The flow is:
1. **Ensure tokenizer exists**
``` python
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.
2. **Compute global latest event timestamp per entity**
``` python
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`.
3. **Create list of tokenized source parts**
``` python
parts = []
```
Everything tokenized gets appended here before concatenation.
4. **Tokenize profile source, if present**
``` python
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 = -1`
- `event_idx = -1`
Inside `_tokenize`, profile fields get `logsec=0` for normal profile
fields, while lifelong fields get `logsec` based on distance from
`eval_time`.
5. **Tokenize each event source**
``` python
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 event
- `hour`
- `dow`
- `dom`
- existing `event_idx`
- key/value ids from the tokenizer
Each source also gets its own `source_idx`.
6. **Concatenate all tokenized sources**
``` python
lf = pl.concat(parts, how='diagonal_relaxed')
```
`diagonal_relaxed` allows profile and event sources to have different
columns. Missing columns become null.
7. **Assign entity shard**
``` python
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.
8. **Fill event time feature nulls**
``` python
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`.
9. **Sort the combined lazy frame**
``` python
.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`
10. **Write one parquet file per shard**
``` python
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:
``` python
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`.
11. **Return output path**
``` python
return self.out_path
```
So 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.
------------------------------------------------------------------------
source
### PRAGMADataset.write_kv
``` python
def write_kv(
eval_time, n_shards:int=100
):
```
*Tokenize sources and write to entity-sharded parquet files.*
``` python
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: none
``` python
out = 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:00source
### PRAGMADataset.show_summary
``` python
def show_summary():
```
*Call self as a function.*
``` python
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 |
``` python
from nbdev.export import nb_export
```
``` python
nb_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.
``` python
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
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 |
``` python
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 |
``` python
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 │
└─────────┴───────────┴───────────┴───────────┴──────┴─────┴──────┘)
``` python
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)
```