01-DataPreprocess.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "cde1550a-36a9-4cba-b681-c91d43a3fe85",
   "metadata": {},
   "outputs": [],
   "source": [
    "%load_ext autoreload\n",
    "%autoreload 2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "32cb87d3-32b2-4051-8bd3-00bf3b4744a6",
   "metadata": {},
   "outputs": [],
   "source": [
    "from data import load_parquets_from_zip\n",
    "from tqdm.auto import tqdm\n",
    "from torch.utils.data import Dataset\n",
    "from functools import partial\n",
    "from itertools import chain\n",
    "import polars as pl\n",
    "import numpy as np\n",
    "import os\n",
    "from collections import defaultdict, Counter\n",
    "from typing import Tuple, Dict, List, Any, Callable, Optional, Union, Iterable\n",
    "from datetime import datetime\n",
    "from bisect import bisect\n",
    "import math\n",
    "import sentence_transformers as st\n",
    "\n",
    "\n",
    "base_dir = 'preprocess'\n",
    "if not os.path.exists(base_dir):\n",
    "    os.makedirs(base_dir)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "c019b0d8-f792-4071-a8c3-abd07bff24e7",
   "metadata": {},
   "outputs": [],
   "source": [
    "from collections import defaultdict\n",
    "import numpy as np\n",
    "def binary_encoding(x: bool) -> int:\n",
    "    return 1 if x else 0\n",
    "\n",
    "\n",
    "def dict_encoding(x: Any, map: Dict[Any, int]) -> int:\n",
    "    return map[x]\n",
    "\n",
    "\n",
    "def build_dict_encoding(labels: List[Any], unknown: bool=False) -> Callable[[Any], int]:\n",
    "    map_elems = {label: i for i, label in enumerate(labels)}\n",
    "    if unknown:\n",
    "        unk_val = len(labels)\n",
    "        map_elems = defaultdict(lambda: unk_val, map_elems)\n",
    "    return partial(dict_encoding, map=map_elems)\n",
    "\n",
    "\n",
    "#Given a list it returns a list with the x quantile limits minus the last limit\n",
    "def quantile_limits(x: Iterable[int], quantiles: int=100) -> List[float]:\n",
    "    data = list(x)\n",
    "    data.sort()\n",
    "    step = len(data) / 100\n",
    "    idxs = [math.ceil(i * step) for i in range(1, 100)]\n",
    "    res = [data[i] for i in idxs]\n",
    "    return res\n",
    "\n",
    "\n",
    "def time_encoding(x: int, limits: List[int]) -> Tuple[int]:\n",
    "    return bisect(limits, x)\n",
    "\n",
    "\n",
    "def count_in_list(df: pl.DataFrame, col: str) -> Dict[Any, int]:\n",
    "    count = Counter()\n",
    "    for x in df[col]:\n",
    "        if x is None:\n",
    "            continue\n",
    "        for v in x:\n",
    "            count[v] += 1\n",
    "    return count\n",
    "\n",
    "\n",
    "def count(df: pl.DataFrame, col: str) -> Dict[Any, int]:\n",
    "    count = Counter()\n",
    "    for x in df[col]:\n",
    "        count[x] += 1\n",
    "    return count\n",
    "\n",
    "\n",
    "def get_map_for_feature(df: pl.DataFrame, col: str, min_reps: int, is_list_column: bool=False, unknown: bool=False) -> Tuple[Callable[[Any], int], List[Any]]:\n",
    "    if is_list_column:\n",
    "        values_count = count_in_list(df, col)\n",
    "    else:\n",
    "        values_count = count(df, col)\n",
    "    filtered_values = [x for x, v in values_count.items() if v > min_reps]\n",
    "    filtered_values.sort()\n",
    "    return build_dict_encoding(filtered_values, unknown=unknown), filtered_values\n",
    "\n",
    "\n",
    "def preprocess_article(article_df: pl.DataFrame, min_reps: int=100, \n",
    "                       image_ids: Optional[Dict[int, int]]=None, \n",
    "                       model_name: str='paraphrase-multilingual-mpnet-base-v2') -> Tuple[pl.DataFrame, pl.DataFrame, pl.DataFrame, pl.DataFrame, Dict[int, int], Dict[int, datetime]]:\n",
    "    ops = []\n",
    "    model = st.SentenceTransformer(model_name)\n",
    "    categories = list(set(article_df['category_str']))\n",
    "    categories.sort()\n",
    "    categories_map = {c: i for i, c in enumerate(categories)}\n",
    "    embeddings = model.encode(categories)\n",
    "    categories_df = pl.DataFrame(data=[categories, embeddings], schema=['category_str', 'embeddings'])\n",
    "    ops.append(pl.col('category_str').replace(categories_map).cast(pl.Int64).alias('category_link'))\n",
    "\n",
    "    ners = set()\n",
    "    for n in article_df['ner_clusters']:\n",
    "        if n is not None:\n",
    "            for w in n:\n",
    "                ners.add(w)\n",
    "    ners = list(ners)\n",
    "    ners.sort()\n",
    "    ner_map = {n: i for i, n in enumerate(ners)}\n",
    "    embeddings = model.encode(ners)\n",
    "    ner_df = pl.DataFrame(data=[ners, embeddings], schema=['ners', 'embeddings'])\n",
    "    ops.append(pl.col('ner_clusters').map_elements(lambda x: [ner_map[n] for n in x] if x is not None else [], \n",
    "                                               return_dtype=pl.List(pl.Int64), skip_nulls=False).alias('ner_clusters_link'))\n",
    "\n",
    "    topics = set()\n",
    "    for n in article_df['topics']:\n",
    "        if n is not None:\n",
    "            for w in n:\n",
    "                topics.add(w)\n",
    "    topics = list(topics)\n",
    "    topics.sort()\n",
    "    topics_map = {n: i for i, n in enumerate(topics)}\n",
    "    embeddings = model.encode(topics)\n",
    "    topics_df = pl.DataFrame(data=[topics, embeddings], schema=['topics', 'embeddings'])\n",
    "    ops.append(pl.col('topics').map_elements(lambda x: [topics_map[n] for n in x] if x is not None else [], \n",
    "                                               return_dtype=pl.List(pl.Int64), skip_nulls=False).alias('topics_link'))\n",
    "\n",
    "    article_df = article_df.with_columns(ops)\n",
    "\n",
    "    #Preprocessing\n",
    "    article_df = article_df.with_columns(pl.col('ner_clusters').map_elements(lambda x: [n.lower()[:3] for n in x] if x is not None else [], \n",
    "                                               return_dtype=pl.List(pl.String), skip_nulls=False))\n",
    "\n",
    "    articleid_date = article_df.select('article_id', 'published_time').rows_by_key(key='article_id')\n",
    "    articleid_date = {k: v[0] for k, v in articleid_date.items()}\n",
    "    articleid_idx = {a: i for i, a in enumerate(article_df['article_id'])}\n",
    "    #Ops to execute on the article_df\n",
    "    ops = [pl.col('premium').map_elements(binary_encoding, return_dtype=pl.Int64),\n",
    "           (pl.col('published_time').dt.weekday() - 1).alias('pub_weekday'),\n",
    "           pl.col('published_time').dt.hour().alias('pub_hour')]\n",
    "    #Is list & unknown\n",
    "    features = [('article_type', False, True),\n",
    "                ('subcategory', True, False),\n",
    "                ('sentiment_label', False, False),\n",
    "                ('category', False, True),\n",
    "                ('ner_clusters', True, False),\n",
    "                ('topics', True, False)]\n",
    "    for feature, is_list, unknown in features:\n",
    "        map_feature, valid_value = get_map_for_feature(article_df, feature, min_reps, is_list_column=is_list, unknown=unknown)\n",
    "        if is_list:\n",
    "            valid_value = set(valid_value)\n",
    "\n",
    "            def process_feature(x, map_feature=map_feature, valid_value=valid_value):\n",
    "                if x is None:\n",
    "                    return []\n",
    "                return [map_feature(v) for v in x if v in valid_value]\n",
    "            \n",
    "            ops.append(pl.col(feature).map_elements(process_feature, return_dtype=pl.List(pl.Int64), skip_nulls=False))\n",
    "        else:\n",
    "            ops.append(pl.col(feature).map_elements(map_feature, return_dtype=pl.Int64))\n",
    "        print(f'Feature {feature} encoded. Number of values: {(len(valid_value) + 1) if unknown else len(valid_value)}')\n",
    "    if image_ids is not None:\n",
    "        ops.append(pl.col('image_ids').map_elements(lambda x: [image_ids[v] for v in x if v in image_ids] if x is not None else [], \n",
    "                                                    return_dtype=pl.List(pl.Int64), skip_nulls=False))\n",
    "    article_df = article_df.with_columns(ops)\n",
    "\n",
    "    return article_df, categories_df, ner_df, topics_df, articleid_idx, articleid_date\n",
    "\n",
    "\n",
    "def compute_behaviors_dates(behavior_df, article_publish):\n",
    "    behavior_df = behavior_df.with_columns(\n",
    "        article_delta_time=pl.struct(('impression_time','article_ids_inview' )). \\\n",
    "                            map_elements(\n",
    "                                lambda x: [int((x['impression_time'] - article_publish[a]).total_seconds()) for a in x['article_ids_inview']],\n",
    "                                return_dtype=pl.List(pl.Int64)),\n",
    "        number_articles=pl.col('article_ids_inview').map_elements(len, return_dtype=pl.Int64)\n",
    "    )\n",
    "    return behavior_df\n",
    "\n",
    "\n",
    "def ids_sort(data: List[Any]) -> List[int]:\n",
    "    idx = list(range(len(data)))\n",
    "    idx.sort(key=lambda x: data[x], reverse=False)\n",
    "    return idx\n",
    "\n",
    "\n",
    "def sort_ids(info: List[Any], idxs: List[int]) -> List[Any]:\n",
    "    return [info[idx] for idx in idxs]\n",
    "\n",
    "\n",
    "def compute_history_dates(history_df: pl.DataFrame, article_publish: Dict[int, datetime], sorted_history: bool=False) -> pl.DataFrame:\n",
    "    if sorted_history:\n",
    "        history_df = history_df.\\\n",
    "                    with_columns(order=pl.col('impression_time_fixed'). \\\n",
    "                                map_elements(ids_sort, return_dtype=pl.List(pl.Int64)))\n",
    "        history_df = history_df.with_columns(\n",
    "                        impression_time_fixed=pl.struct('impression_time_fixed', 'order').\n",
    "                        map_elements(lambda x: sort_ids(x['impression_time_fixed'], x['order']), \n",
    "                                    return_dtype=pl.List(pl.Datetime)),\n",
    "                        scroll_percentage_fixed=pl.struct('scroll_percentage_fixed', 'order').\n",
    "                        map_elements(lambda x: sort_ids(x['scroll_percentage_fixed'], x['order']), \n",
    "                                    return_dtype=pl.List(pl.Float64)),\n",
    "                        article_id_fixed=pl.struct('article_id_fixed', 'order').\n",
    "                        map_elements(lambda x: sort_ids(x['article_id_fixed'], x['order']), \n",
    "                                    return_dtype=pl.List(pl.Int64)),\n",
    "                        read_time_fixed=pl.struct('read_time_fixed', 'order').\n",
    "                        map_elements(lambda x: sort_ids(x['read_time_fixed'], x['order']), \n",
    "                                    return_dtype=pl.List(pl.Float64))). \\\n",
    "                        drop('order')\n",
    "    history_df = history_df.with_columns(\n",
    "        impression_weekday=pl.col('impression_time_fixed'). \n",
    "            map_elements(lambda x: [d.weekday() for d in x], return_dtype=pl.List(pl.Int64)),\n",
    "        impression_hour=pl.col('impression_time_fixed'). \n",
    "            map_elements(lambda x: [d.hour for d in x], return_dtype=pl.List(pl.Int64)),\n",
    "        article_delta_time=pl.struct(('impression_time_fixed', 'article_id_fixed')). \n",
    "            map_elements(lambda x:[int((d - article_publish[a]).total_seconds()) for d, a in zip(x['impression_time_fixed'], x['article_id_fixed'])], \n",
    "                         return_dtype=pl.List(pl.Int64)),\n",
    "        number_articles=pl.col('article_id_fixed').map_elements(len, return_dtype=pl.Int64)\n",
    "    )\n",
    "    return history_df\n",
    "\n",
    "\n",
    "def preprocess(behavior: pl.DataFrame, history: pl.DataFrame, article_publish: Dict[int, datetime], \n",
    "               article_idx: Dict[int, int], sorted_history: bool=False, time_quartiles: Optional[List[int]]=None) -> Tuple[pl.DataFrame, pl.DataFrame, List[int]]:\n",
    "    print('Computing history & behavior dates')\n",
    "    history = compute_history_dates(history, article_publish, sorted_history)\n",
    "    behavior = compute_behaviors_dates(behavior, article_publish)\n",
    "    \n",
    "    print('Computing history user_id mapping and time percentiles')\n",
    "    historyid_idx = {a: i for i, a in enumerate(history['user_id'])}\n",
    "    if time_quartiles is None:\n",
    "        time_quartiles = quantile_limits(chain(*history['article_delta_time'].to_list()))\n",
    "\n",
    "    print('Preprocessing behavior')\n",
    "    if 'next_scroll_percentage' in behavior.columns:\n",
    "        behavior = behavior.with_columns(pl.col('next_scroll_percentage').\n",
    "                                         map_elements(lambda x: x / 100 if x is not None else 0.0, \n",
    "                                                      return_dtype=pl.Float64, skip_nulls=False),\n",
    "                                        article_ids_clicked=pl.struct('article_ids_inview', 'article_ids_clicked').\n",
    "                                        map_elements(lambda x: [x['article_ids_inview'].index(v) for v in x['article_ids_clicked']], \n",
    "                                                     return_dtype=pl.List(pl.Int64)))\n",
    "    behavior = behavior.with_columns(\n",
    "        pl.col('user_id').replace(historyid_idx),\n",
    "        pl.col('is_sso_user').map_elements(binary_encoding, return_dtype=pl.Int8),\n",
    "        pl.col('gender').map_elements(build_dict_encoding([0, 1, 2, None]), return_dtype=pl.Int64, skip_nulls=False),\n",
    "        pl.col('postcode').map_elements(build_dict_encoding([0, 1, 2, 3, 4, None]), return_dtype=pl.Int64, skip_nulls=False),\n",
    "        pl.col('age').map_elements(build_dict_encoding([0, 10, 20, 30, 40, 50, 60, 70, 80, 90, 100, None]), return_dtype=pl.Int64, skip_nulls=False),\n",
    "        pl.col('is_subscriber').map_elements(binary_encoding, return_dtype=pl.Int8),\n",
    "        pl.col('device_type').map_elements(build_dict_encoding([0, 1, 2, 3]), return_dtype=pl.Int8),\n",
    "        pl.col('article_delta_time').map_elements(lambda x: [time_encoding(v, time_quartiles) for v in x], return_dtype=pl.List(pl.Int64)),\n",
    "        pl.col('article_ids_inview').map_elements(lambda x: [article_idx[v] for v in x], return_dtype=pl.List(pl.Int64)),\n",
    "        impression_hour=pl.col('impression_time').dt.hour(),\n",
    "        impression_weekday=pl.col('impression_time').dt.weekday() - 1\n",
    "    )\n",
    "    print('Preprocessing history')\n",
    "    history = history.with_columns(pl.col('scroll_percentage_fixed').\n",
    "                                   map_elements(lambda l: [x / 100 if x != None else 0.0 for x in l], \n",
    "                                                return_dtype=pl.List(pl.Float64)),\n",
    "                                    pl.col('article_delta_time').map_elements(lambda x: [time_encoding(v, time_quartiles) for v in x], return_dtype=pl.List(pl.Int64)),\n",
    "                                    pl.col('article_id_fixed').map_elements(lambda x: [article_idx[v] for v in x], return_dtype=pl.List(pl.Int64))\n",
    "                                    )\n",
    "    return behavior, history, time_quartiles"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "ac1776e4-bf73-49ba-903d-e715245d7380",
   "metadata": {},
   "outputs": [],
   "source": [
    "ds = load_parquets_from_zip('dataset/ebnerd_large.zip')\n",
    "behavior = ds['train/behaviors']\n",
    "history = ds['train/history']\n",
    "article = ds['articles']\n",
    "del ds"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "ba8432e9",
   "metadata": {},
   "outputs": [],
   "source": [
    "# images = load_parquets_from_zip('dataset/Ekstra_Bladet_image_embeddings.zip')['Ekstra_Bladet_image_embeddings/image_embeddings']\n",
    "# image_ids = {a: i for i, a in enumerate(images['article_id'])}\n",
    "# del images\n",
    "# preprocess_article(article, image_ids=image_ids)[0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "5197b771",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Feature article_type encoded. Number of values: 4\n",
      "Feature subcategory encoded. Number of values: 28\n",
      "Feature sentiment_label encoded. Number of values: 3\n",
      "Feature category encoded. Number of values: 14\n",
      "Feature ner_clusters encoded. Number of values: 96\n",
      "Feature topics encoded. Number of values: 54\n"
     ]
    }
   ],
   "source": [
    "article, categories_embs, ner_embs, topics_embs, articleid_idx, article_publish_date = preprocess_article(article, 1000)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "d994ea18",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Computing history & behavior dates\n",
      "Computing history user_id mapping and time percentiles\n",
      "Preprocessing behavior\n",
      "Preprocessing history\n",
      "CPU times: total: 8min 43s\n",
      "Wall time: 22min 45s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "behavior, history, time_quartiles = preprocess(behavior, history, article_publish_date, articleid_idx)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "9d4abc33-f108-4996-97d7-0f874b845f15",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div><style>\n",
       ".dataframe > thead > tr,\n",
       ".dataframe > tbody > tr {\n",
       "  text-align: right;\n",
       "  white-space: pre-wrap;\n",
       "}\n",
       "</style>\n",
       "<small>shape: (5, 21)</small><table border=\"1\" class=\"dataframe\"><thead><tr><th>impression_id</th><th>article_id</th><th>impression_time</th><th>read_time</th><th>scroll_percentage</th><th>device_type</th><th>article_ids_inview</th><th>article_ids_clicked</th><th>user_id</th><th>is_sso_user</th><th>gender</th><th>postcode</th><th>age</th><th>is_subscriber</th><th>session_id</th><th>next_read_time</th><th>next_scroll_percentage</th><th>article_delta_time</th><th>number_articles</th><th>impression_hour</th><th>impression_weekday</th></tr><tr><td>u32</td><td>i32</td><td>datetime[μs]</td><td>f32</td><td>f32</td><td>i8</td><td>list[i64]</td><td>list[i64]</td><td>i64</td><td>i8</td><td>i64</td><td>i64</td><td>i64</td><td>i8</td><td>u32</td><td>f32</td><td>f64</td><td>list[i64]</td><td>i64</td><td>i8</td><td>i8</td></tr></thead><tbody><tr><td>47727</td><td>null</td><td>2023-05-21 21:35:07</td><td>20.0</td><td>null</td><td>1</td><td>[95773, 122630, … 99525]</td><td>[1]</td><td>341731</td><td>0</td><td>3</td><td>5</td><td>11</td><td>0</td><td>265</td><td>34.0</td><td>1.0</td><td>[99, 75, … 99]</td><td>6</td><td>21</td><td>6</td></tr><tr><td>47731</td><td>null</td><td>2023-05-21 21:32:33</td><td>13.0</td><td>null</td><td>1</td><td>[122562, 122550, … 120591]</td><td>[4]</td><td>341731</td><td>0</td><td>3</td><td>5</td><td>11</td><td>0</td><td>265</td><td>45.0</td><td>1.0</td><td>[49, 62, … 61]</td><td>5</td><td>21</td><td>6</td></tr><tr><td>47736</td><td>null</td><td>2023-05-21 21:33:32</td><td>17.0</td><td>null</td><td>1</td><td>[120591, 122562, … 122638]</td><td>[9]</td><td>341731</td><td>0</td><td>3</td><td>5</td><td>11</td><td>0</td><td>265</td><td>78.0</td><td>1.0</td><td>[62, 49, … 63]</td><td>13</td><td>21</td><td>6</td></tr><tr><td>47737</td><td>null</td><td>2023-05-21 21:38:17</td><td>27.0</td><td>null</td><td>1</td><td>[122567, 122626, … 122602]</td><td>[9]</td><td>341731</td><td>0</td><td>3</td><td>5</td><td>11</td><td>0</td><td>265</td><td>6.0</td><td>0.52</td><td>[81, 76, … 82]</td><td>11</td><td>21</td><td>6</td></tr><tr><td>47740</td><td>null</td><td>2023-05-21 21:36:02</td><td>48.0</td><td>null</td><td>1</td><td>[122595, 122629, … 122578]</td><td>[8]</td><td>341731</td><td>0</td><td>3</td><td>5</td><td>11</td><td>0</td><td>265</td><td>32.0</td><td>1.0</td><td>[73, 72, … 76]</td><td>9</td><td>21</td><td>6</td></tr></tbody></table></div>"
      ],
      "text/plain": [
       "shape: (5, 21)\n",
       "┌───────────────┬────────────┬─────────────────────┬───────────┬───┬────────────────────┬─────────────────┬─────────────────┬────────────────────┐\n",
       "│ impression_id ┆ article_id ┆ impression_time     ┆ read_time ┆ … ┆ article_delta_time ┆ number_articles ┆ impression_hour ┆ impression_weekday │\n",
       "│ ---           ┆ ---        ┆ ---                 ┆ ---       ┆   ┆ ---                ┆ ---             ┆ ---             ┆ ---                │\n",
       "│ u32           ┆ i32        ┆ datetime[μs]        ┆ f32       ┆   ┆ list[i64]          ┆ i64             ┆ i8              ┆ i8                 │\n",
       "╞═══════════════╪════════════╪═════════════════════╪═══════════╪═══╪════════════════════╪═════════════════╪═════════════════╪════════════════════╡\n",
       "│ 47727         ┆ null       ┆ 2023-05-21 21:35:07 ┆ 20.0      ┆ … ┆ [99, 75, … 99]     ┆ 6               ┆ 21              ┆ 6                  │\n",
       "│ 47731         ┆ null       ┆ 2023-05-21 21:32:33 ┆ 13.0      ┆ … ┆ [49, 62, … 61]     ┆ 5               ┆ 21              ┆ 6                  │\n",
       "│ 47736         ┆ null       ┆ 2023-05-21 21:33:32 ┆ 17.0      ┆ … ┆ [62, 49, … 63]     ┆ 13              ┆ 21              ┆ 6                  │\n",
       "│ 47737         ┆ null       ┆ 2023-05-21 21:38:17 ┆ 27.0      ┆ … ┆ [81, 76, … 82]     ┆ 11              ┆ 21              ┆ 6                  │\n",
       "│ 47740         ┆ null       ┆ 2023-05-21 21:36:02 ┆ 48.0      ┆ … ┆ [73, 72, … 76]     ┆ 9               ┆ 21              ┆ 6                  │\n",
       "└───────────────┴────────────┴─────────────────────┴───────────┴───┴────────────────────┴─────────────────┴─────────────────┴────────────────────┘"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "behavior.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "27f8f187-538a-4843-a468-4113fb5c145a",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div><style>\n",
       ".dataframe > thead > tr,\n",
       ".dataframe > tbody > tr {\n",
       "  text-align: right;\n",
       "  white-space: pre-wrap;\n",
       "}\n",
       "</style>\n",
       "<small>shape: (5, 9)</small><table border=\"1\" class=\"dataframe\"><thead><tr><th>user_id</th><th>impression_time_fixed</th><th>scroll_percentage_fixed</th><th>article_id_fixed</th><th>read_time_fixed</th><th>impression_weekday</th><th>impression_hour</th><th>article_delta_time</th><th>number_articles</th></tr><tr><td>u32</td><td>list[datetime[μs]]</td><td>list[f64]</td><td>list[i64]</td><td>list[f32]</td><td>list[i64]</td><td>list[i64]</td><td>list[i64]</td><td>i64</td></tr></thead><tbody><tr><td>10029</td><td>[2023-04-28 06:16:57, 2023-04-28 06:17:31, … 2023-05-18 06:59:50]</td><td>[0.23, 0.69, … 0.0]</td><td>[117123, 117715, … 122002]</td><td>[28.0, 24.0, … 0.0]</td><td>[4, 4, … 3]</td><td>[6, 6, … 6]</td><td>[0, 18, … 67]</td><td>678</td></tr><tr><td>10033</td><td>[2023-04-27 11:11:32, 2023-04-27 11:12:56, … 2023-05-17 20:22:42]</td><td>[0.33, 0.41, … 0.29]</td><td>[117462, 117475, … 121840]</td><td>[2.0, 2.0, … 1.0]</td><td>[3, 3, … 2]</td><td>[11, 11, … 20]</td><td>[81, 76, … 29]</td><td>587</td></tr><tr><td>10034</td><td>[2023-04-30 09:46:57, 2023-04-30 09:47:33, … 2023-05-16 08:40:52]</td><td>[0.0, 0.88, … 1.0]</td><td>[118112, 118110, … 121595]</td><td>[21.0, 103.0, … 9.0]</td><td>[6, 6, … 1]</td><td>[9, 9, … 8]</td><td>[69, 59, … 33]</td><td>140</td></tr><tr><td>10041</td><td>[2023-04-27 15:15:28, 2023-04-27 15:16:30, … 2023-05-17 14:54:05]</td><td>[0.78, 0.41, … 0.57]</td><td>[117552, 117478, … 120316]</td><td>[12.0, 11.0, … 22.0]</td><td>[3, 3, … 2]</td><td>[15, 15, … 14]</td><td>[39, 62, … 77]</td><td>139</td></tr><tr><td>10103</td><td>[2023-04-27 15:37:35, 2023-04-27 15:38:37, … 2023-05-18 04:52:09]</td><td>[1.0, 0.0, … 0.63]</td><td>[117552, 117571, … 121846]</td><td>[45.0, 8.0, … 24.0]</td><td>[3, 3, … 3]</td><td>[15, 15, … 4]</td><td>[49, 3, … 92]</td><td>64</td></tr></tbody></table></div>"
      ],
      "text/plain": [
       "shape: (5, 9)\n",
       "┌─────────┬───────────────────────┬─────────────────────────┬─────────────────────────────┬───┬────────────────────┬─────────────────┬────────────────────┬─────────────────┐\n",
       "│ user_id ┆ impression_time_fixed ┆ scroll_percentage_fixed ┆ article_id_fixed            ┆ … ┆ impression_weekday ┆ impression_hour ┆ article_delta_time ┆ number_articles │\n",
       "│ ---     ┆ ---                   ┆ ---                     ┆ ---                         ┆   ┆ ---                ┆ ---             ┆ ---                ┆ ---             │\n",
       "│ u32     ┆ list[datetime[μs]]    ┆ list[f64]               ┆ list[i64]                   ┆   ┆ list[i64]          ┆ list[i64]       ┆ list[i64]          ┆ i64             │\n",
       "╞═════════╪═══════════════════════╪═════════════════════════╪═════════════════════════════╪═══╪════════════════════╪═════════════════╪════════════════════╪═════════════════╡\n",
       "│ 10029   ┆ [2023-04-28 06:16:57, ┆ [0.23, 0.69, … 0.0]     ┆ [117123, 117715, … 122002]  ┆ … ┆ [4, 4, … 3]        ┆ [6, 6, … 6]     ┆ [0, 18, … 67]      ┆ 678             │\n",
       "│         ┆ 2023-04-28…           ┆                         ┆                             ┆   ┆                    ┆                 ┆                    ┆                 │\n",
       "│ 10033   ┆ [2023-04-27 11:11:32, ┆ [0.33, 0.41, … 0.29]    ┆ [117462, 117475, … 121840]  ┆ … ┆ [3, 3, … 2]        ┆ [11, 11, … 20]  ┆ [81, 76, … 29]     ┆ 587             │\n",
       "│         ┆ 2023-04-27…           ┆                         ┆                             ┆   ┆                    ┆                 ┆                    ┆                 │\n",
       "│ 10034   ┆ [2023-04-30 09:46:57, ┆ [0.0, 0.88, … 1.0]      ┆ [118112, 118110, … 121595]  ┆ … ┆ [6, 6, … 1]        ┆ [9, 9, … 8]     ┆ [69, 59, … 33]     ┆ 140             │\n",
       "│         ┆ 2023-04-30…           ┆                         ┆                             ┆   ┆                    ┆                 ┆                    ┆                 │\n",
       "│ 10041   ┆ [2023-04-27 15:15:28, ┆ [0.78, 0.41, … 0.57]    ┆ [117552, 117478, … 120316]  ┆ … ┆ [3, 3, … 2]        ┆ [15, 15, … 14]  ┆ [39, 62, … 77]     ┆ 139             │\n",
       "│         ┆ 2023-04-27…           ┆                         ┆                             ┆   ┆                    ┆                 ┆                    ┆                 │\n",
       "│ 10103   ┆ [2023-04-27 15:37:35, ┆ [1.0, 0.0, … 0.63]      ┆ [117552, 117571, … 121846]  ┆ … ┆ [3, 3, … 3]        ┆ [15, 15, … 4]   ┆ [49, 3, … 92]      ┆ 64              │\n",
       "│         ┆ 2023-04-27…           ┆                         ┆                             ┆   ┆                    ┆                 ┆                    ┆                 │\n",
       "└─────────┴───────────────────────┴─────────────────────────┴─────────────────────────────┴───┴────────────────────┴─────────────────┴────────────────────┴─────────────────┘"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "history.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "f330299c-5e31-49f7-9411-d5384b9652ad",
   "metadata": {},
   "outputs": [],
   "source": [
    "behavior.write_parquet('preprocess/train_behaviors.parquet')\n",
    "history.write_parquet('preprocess/train_history.parquet')\n",
    "article.write_parquet('preprocess/article.parquet')\n",
    "categories_embs.write_parquet('preprocess/categories_embs.parquet')\n",
    "ner_embs.write_parquet('preprocess/ner_embs.parquet')\n",
    "topics_embs.write_parquet('preprocess/topics_embs.parquet')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "30807062",
   "metadata": {},
   "outputs": [],
   "source": [
    "ds = load_parquets_from_zip('dataset/ebnerd_large.zip')\n",
    "behavior = ds['validation/behaviors']\n",
    "history = ds['validation/history']\n",
    "del ds"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "e7293f81",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Computing history & behavior dates\n",
      "Computing history user_id mapping and time percentiles\n",
      "Preprocessing behavior\n",
      "Preprocessing history\n",
      "CPU times: total: 6min 57s\n",
      "Wall time: 21min 50s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "behavior, history, _ = preprocess(behavior, history, article_publish_date, articleid_idx, time_quartiles=time_quartiles)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "92f45e4a",
   "metadata": {},
   "outputs": [],
   "source": [
    "behavior.write_parquet('preprocess/validation_behaviors.parquet')\n",
    "history.write_parquet('preprocess/validation_history.parquet')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "7cbb3033",
   "metadata": {},
   "outputs": [],
   "source": [
    "del behavior\n",
    "del history"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "6a08c291-2b24-4758-820c-1ac0d30d5858",
   "metadata": {},
   "outputs": [],
   "source": [
    "ds = load_parquets_from_zip('dataset/ebnerd_testset.zip')\n",
    "behavior = ds['ebnerd_testset/test/behaviors']\n",
    "history = ds['ebnerd_testset/test/history']\n",
    "del ds"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "7d76f268-b824-4e13-a08d-ab35a6d8684a",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Computing history & behavior dates\n",
      "Computing history user_id mapping and time percentiles\n",
      "Preprocessing behavior\n",
      "Preprocessing history\n",
      "CPU times: total: 8min 6s\n",
      "Wall time: 15min 12s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "behavior, history, _ = preprocess(behavior, history, article_publish_date, articleid_idx, time_quartiles=time_quartiles)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "68a83bbc-9530-4b0e-adfd-6ffa31b7846d",
   "metadata": {},
   "outputs": [],
   "source": [
    "behavior.write_parquet('preprocess/test_behaviors.parquet')\n",
    "history.write_parquet('preprocess/test_history.parquet')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "6a58e055-7eb3-4a67-ab2b-1510ee2a8701",
   "metadata": {},
   "outputs": [],
   "source": [
    "del behavior\n",
    "del history"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}