{"cells":[{"metadata":{"id":"N4lHWTOFrRgT","_kg_hide-input":true,"trusted":false},"cell_type":"code","source":"# \ndef show_version_history():\n from IPython.display import HTML\n style_header = 'mui--align-bottom mui--bg-primary mui--text-light mui--text-center'\n style_cell = 'mui--align-top mui--text-center'\n\n # print(('\\n').join(list(map(make_li, sorted(dtypes.keys())))))\n\n html_str = f\"\"\"\n \n
\n

Version History

\n
\n
\n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n
VersionDateLocal CVPublic
Leaderboard		Notes
602020-11-150.7560.762
    \n
  • Completed submission pipeline with minimal feature set:\n
      \n
    • answered_correctly_content_id_cumsum
      • \n
    • answered_correctly_cumsum
      • \n
    • answered_correctly_cumsum_pct
      • \n
    • answered_incorrectly_content_id_cumsum
      • \n
    • answered_incorrectly_cumsum
      • \n
    • part
      • \n
    • part_correct_pct
      • \n
    • question_id_correct_pct
      • \n
    • tag__0
      • \n
    • tag__0_correct_pct
      • \n
    • task_container_id
      • \n
    • timestamp
      • \n
    \n
    • \n
  • Changed logic on roll sum to be over trailing\n rows preceding the current task_container_id instead\n of over trailing task containers\n (expensive)\n
    • \n
\n
532020-11-070.761--
  • Housekeeping:\n
      \n
    • Consolidated notebook and modules in single repo
      • \n
    • Streamlined Colab repo workflow using Drive\n
    • Included modules in notebook when pushed to Kaggle
      • \n
    • Eliminated CONFIG requirement when run in Kaggle
      • \n
    \n
    • \n
\n
402020-11-050.761--\n
    \n
  • Features added:\n
      \n
    • answered_correctly_content_id_cumsum
      • \n
    • answered_correctly_content_id_cumsum_pct
      • \n
    • answered_correctly_cumsum10
      • \n
    • answered_correctly_cumsum_pct
      • \n
    • answered_correctly_rollsum_pct
      • \n
    • answered_incorrectly_content_id_cumsum
      • \n
    • lectures_cumcount
      • \n
    • prior_question_elapsed_time_rollavg
      • \n
    \n
    • \n
  • Single model, single fold
    • \n
  • No public leaderboard - efficient inference in progress
    • \n
  • Refactored code to move queries and helper functions into\n separate modules
    • \n
  • Completed set up to commit code to Github from Colab and
    • \n
  • Completed set up to push kernels to Kaggle from Colab
    • \n
\n
372020-11-040.7510.748\n
    \n
  • Features added:\n
      \n
    • answered_correctly_cumsum
      • \n
    • answered_correctly_rollsum
      • \n
    • answered_incorrectly_cumsum
      • \n
    • answered_incorrectly_rollsum
      • \n
    • part
      • \n
    • part_correct_pct
      • \n
    • question_id_correct_pct
      • \n
    • tag__0
      • \n
    • tag__0_correct_pct
      • \n
    \n
    • \n
  • Single model, single fold
    • \n
  • Model for public leaderboard didn't include\n rolling features - still working out how to\n efficiently calculate for inference
    • \n
\n
\n
\n
\n

\n \n \"Open\n

\n
\n \"\"\"\n\n html = HTML(html_str)\n display(html)\nshow_version_history()","execution_count":null,"outputs":[]},{"metadata":{"id":"8QCO0cr_juQA"},"cell_type":"markdown","source":"## Intro"},{"metadata":{"id":"L1mQ0DiRnfer"},"cell_type":"markdown","source":"This kernel is an end to end pipeline that uses BigQuery to store data and perform feature engineering, and trains a model using XGBoost. I was resorting to breaking up tables and still waiting a long time to see the results of my analysis and to process my engineered features, so I decided to learn about BigQuery. This kernel is the current state of my setup, which is working very well. It is much faster than my previous local setup, even with having to download files. It also is making it easier to keep the structure of the data and and code clean, which in turn makes it easier to stay focused on thinking about and executing ideas without getting bogged down waiting for things to finish or wading through extraneous processing code.\n\nI've attempted to put this book together in such a way that somebody else can fork it, update a few environment variables, run it and then be in the game engineering features and improving the model. The only requirements are a GCP project and storage bucket. Other than that, it is turn key, starting with creating a BigQuery dataset and ending with a saved model and two feature tables that get uploaded to a Kaggle dataset where they are used in a separate kernel to make predictions and submit to the competition api.\n\nA couple of cool features:\n* Uses the gcs version of the competition datset to create a dataset and upload to BigQuery in around a minute\n* Transformations get run on the entire train table at once and run in under 10 minutes\n* Feature engineering gets done on a sample of the train table, taking advantage of BigQuery' graphical query editing interface that includes tab completion, syntax checking and the ability to run queries and inspect results\n* Stores queries as methods on a dedicated class, where they can be easily reused\n* Dtypes for local dataframes, schema for BigQuery tables and all tranformations are maintained locally so that the transformed tables can be recreated from the original competition dataset files automatically at any time (see description of workflow below to continue with this practice)\n* Exports to gcs using temporary tables created by BigQuery avoiding unnecessary storage and wasted time rerunning and exporting duplicate queries\n* Separate [submission kernel](https://www.kaggle.com/calebeverett/riiid-submit) uses sqlite3 to achieve sub two hour submission times while maintaining state for questions, users and user-content (80+ million rows)\n\nI've engineered a few features as a starting point to demonstrate how additional features can be efficiently developed and processed, including:\n* Cumulative and rolling sums of questions answered correctly and incorrectly by user\n* Percent of questions answered correctly by question id, part and the first question tag\n\nThe model is also just a starting point, with a first pass at a train/validation split and no hyperparameter tuning. I have included some basic diagnostics on both the train/validtion split and model performance as a starting place for further development.\n\nI have the table creation and transformation functions set to not run, but you can set them to run, by changing the flags to `True` for:\n* Loading tables - one flag for the questions table and another for the train and lectures tables\n* Updating the schemas in BigQuery\n* Performing the transformations"},{"metadata":{"id":"savju5R9juQB"},"cell_type":"markdown","source":"## Resources\n* [BigQuery Console](https://console.cloud.google.com/bigquery)\n* [Python Client for Google BigQuery](https://googleapis.dev/python/bigquery/latest/index.html)\n* [Analytic function concepts in Standard SQL](https://cloud.google.com/bigquery/docs/reference/standard-sql/analytic-function-concepts)\n* [XGBoost Documentation](https://xgboost.readthedocs.io/en/latest/index.html)\n* [Storge Client](https://googleapis.dev/python/storage/latest/client.html)\n* [pandas documentation](https://pandas.pydata.org/docs/)\n* [Plotly Python Open Source Graphing Library](https://plotly.com/python/)\n* [PEP 8 -- Style Guide for Python Code](https://www.python.org/dev/peps/pep-0008/)"},{"metadata":{"id":"88t93RZzj982"},"cell_type":"markdown","source":"## Imports"},{"metadata":{"id":"tKe4S7M4nbSU","_kg_hide-input":true,"trusted":false},"cell_type":"code","source":"# \n%load_ext autoreload\n%autoreload 2\n\nfrom datetime import datetime\nimport gc\nimport json\nimport os\nfrom pathlib import Path\nimport re\nimport subprocess\nimport sys\nimport time\n\nimport ipywidgets as widgets\nfrom google.cloud import storage, bigquery\nfrom google.cloud.bigquery import SchemaField\nimport numpy as np\nimport pandas as pd\nfrom tqdm.notebook import tqdm\n\nBUCKET = 'riiid-caleb'\nDATASET = 'data'\nLOCATION = 'us-west4'\nKAGGLE_SUBMIT_DATASET = 'riiid-submission'\nPROJECT = 'riiid-caleb'\nREPO = 'riiid_2020'\nNOT_KAGGLE = os.getenv('KAGGLE_URL_BASE') is None\n\nif NOT_KAGGLE:\n from google.colab import drive\n DRIVE = Path('/content/drive/My Drive')\n if not DRIVE.exists():\n drive.mount(str(DRIVE.parent)) \n sys.path.append(str(DRIVE))\n g_creds_path = 'credentials/riiid-caleb-faddd0c9d900.json'\n os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = str(DRIVE/g_creds_path)\n\nbucket = storage.Client(project=PROJECT).get_bucket(BUCKET)\ndataset = bigquery.Dataset(f'{PROJECT}.{DATASET}')\nbq_client = bigquery.Client(project=PROJECT, location=LOCATION)\n\nif NOT_KAGGLE:\n CONFIG = json.loads(bucket.get_blob('config.json').download_as_string())\n os.environ = {**os.environ, **CONFIG}\n from riiid_2020.utilities import check_packages, Git\n from riiid_2020.bqhelpers import BQHelper\n from riiid_2020.queries import Queries\n \n git = Git(REPO, CONFIG.get('GIT_USERNAME'), CONFIG.get('GIT_PASSWORD'),\n CONFIG.get('EMAIL'), DRIVE)\n\n packages = {\n 'comet-ml': '3.2.5',\n 'gcsfs': '0.7.1',\n 'kaggle': '1.5.9',\n 'plotly': '4.12.0',\n 'xgboost': '1.2.0',\n }\n check_packages(packages)\n\n from comet_ml import Experiment\n from kaggle.api.kaggle_api_extended import KaggleApi\n kaggle_api = KaggleApi()\n kaggle_api.authenticate()\n\nimport plotly\nimport plotly.express as px\nfrom sklearn.metrics import roc_auc_score\nfrom sklearn.preprocessing import MultiLabelBinarizer\nimport xgboost as xgb\npd.options.plotting.backend = 'plotly'","execution_count":null,"outputs":[]},{"metadata":{"id":"3I6-yy4QBxlR"},"cell_type":"markdown","source":"## Modules\nIncluded in notebook for convenience when in a Kaggle kernel. Github repo [here](https://github.com/CalebEverett/riiid_2020)."},{"metadata":{"id":"--uRkD6w0EhS","_kg_hide-input":true,"trusted":false},"cell_type":"code","source":"from pathlib import Path\nimport pytz\nimport sys\n\nfrom google.cloud.bigquery import ExtractJobConfig, LoadJobConfig, \\\n SchemaField, SourceFormat\nimport pandas as pd\nfrom tqdm.notebook import tqdm\n\n\nclass BQHelper:\n def __init__(self, bucket, DATASET, bq_client):\n self.bucket = bucket\n self.BUCKET = self.bucket.name\n self.DATASET = DATASET\n self.bq_client = bq_client\n \n # LOAD FUNCTTIONS \n # ================\n def load_job_cb(self, future):\n \"\"\"Prints update upon completion to output of last run cell.\"\"\"\n \n seconds = (future.ended - future.created).total_seconds()\n print(f'Loaded {future.output_rows:,d} rows to table {future.job_id.split(\"_\")[0]} in '\n f'{seconds:>4,.1f} sec, {int(future.output_rows / seconds):,d} per sec.')\n \n def load_csv_uri(self, table_id, schemas_orig):\n full_table_id = f'{self.DATASET}.{table_id}'\n\n job_config = LoadJobConfig(\n schema=schemas_orig[table_id],\n source_format=SourceFormat.CSV,\n skip_leading_rows=1\n )\n\n uri = f'gs://{self.BUCKET}/{table_id}.csv'\n load_job = self.bq_client.load_table_from_uri(uri, full_table_id,\n job_config=job_config,\n job_id_prefix=f'{table_id}_')\n print(f'job {load_job.job_id} started')\n load_job.add_done_callback(self.load_job_cb)\n \n return load_job\n \n def load_json_file(self, table_id, schemas_orig):\n full_table_id = f'{self.DATASET}.{table_id}'\n\n job_config = LoadJobConfig(\n schema=schemas_orig[table_id],\n source_format=SourceFormat.NEWLINE_DELIMITED_JSON)\n\n file_path = f'{table_id}.json'\n with open(file_path, \"rb\") as source_file:\n load_job = self.bq_client.load_table_from_file(source_file,\n full_table_id,\n job_config=job_config,\n job_id_prefix=f'{table_id}_')\n print(f'job {load_job.job_id} started')\n load_job.add_done_callback(self.load_job_cb)\n \n return load_job\n\n def get_table(self, table_id):\n return self.bq_client.get_table(f'{self.DATASET}.{table_id}')\n\n def del_table(self, table_id):\n return self.bq_client.delete_table(f'{self.DATASET}.{table_id}',\n not_found_ok=True)\n\n def get_df_jobs(self, max_results=10):\n jobs = self.bq_client.list_jobs(max_results=max_results, all_users=True)\n jobs_list = []\n\n if jobs.num_results:\n for job in jobs:\n ended = job.ended if job.ended else datetime.now(pytz.UTC)\n exception = job.exception() if job.ended else None\n jobs_list.append({'job_id': job.job_id, 'job_type': job.job_type,\n 'started': job.started, 'ended': ended,\n 'running': job.running(),\n 'exception': exception,\n })\n df_jobs = pd.DataFrame(jobs_list)\n df_jobs['seconds'] = (df_jobs.ended - df_jobs.started).dt.seconds\n df_jobs.started = df_jobs.started.astype(str).str[:16]\n del df_jobs['ended']\n return df_jobs\n else:\n return None\n\n def get_df_table_list(self):\n tables = []\n for t in self.bq_client.list_tables(self.DATASET):\n table = self.bq_client.get_table(t)\n tables.append({'table_id': table.table_id, 'cols': len(table.schema),\n 'rows': table.num_rows, 'kb': int(table.num_bytes/1e3)})\n df_tables = pd.DataFrame(tables)\n \n return df_tables\n\n # QUERY FUNCTIONS\n # ================\n def done_cb(self, future):\n seconds = (future.ended - future.started).total_seconds()\n print(f'Job {future.job_id} finished in {seconds} seconds.')\n\n def run_query(self, query, job_id_prefix=None, wait=False):\n query_job = self.bq_client.query(query, job_id_prefix=job_id_prefix)\n print(f'Job {query_job.job_id} started.')\n query_job.add_done_callback(self.done_cb)\n if wait:\n query_job.result()\n \n return query_job\n\n def get_df_query(self, query, dtypes=None):\n query_job = self.run_query(*query)\n\n df_query = query_job.to_dataframe(dtypes=dtypes, \n progress_bar_type='tqdm_notebook')\n return df_query\n\n def get_df_table(self, table_id, max_results=10000, dtypes=None):\n table = self.get_table(table_id)\n df_table = (self.bq_client.list_rows(table, max_results=max_results)\n .to_dataframe(dtypes=dtypes,\n progress_bar_type='tqdm_notebook'))\n return df_table\n\n # EXPORT FUNCTIONS\n # ================\n def export_query_gcs(self, query, file_format='csv', wait=True):\n \"\"\" Uses BigQuery temporary table reference as gcs prefix.\n Runs query and exports to gcs if it doesn't already exist in gcs.\n Exported in multiple files if over 1GB. Returns gcs prefix.\n \"\"\"\n qj = self.run_query(*query, wait=wait)\n \n prefix = ('/').join(qj.destination.path.split('/')[-2:])\n blobs_list = list(self.bucket.list_blobs(prefix=prefix))\n \n if not blobs_list:\n \n job_prefix_id = sys._getframe().f_code.co_name + '_'\n \n formats={'csv': 'CSV',\n 'json': 'NEWLINE_DELIMITED_JSON'}\n \n job_config = ExtractJobConfig(destination_format=formats[file_format])\n \n ex_job = self.bq_client.extract_table(\n source=qj.destination,\n destination_uris=f'gs://{self.BUCKET}/{prefix}/*.{file_format}',\n job_id_prefix=job_prefix_id,\n job_config=job_config)\n \n ex_job.add_done_callback(self.done_cb)\n \n print(f'Job {ex_job.job_id} started.')\n \n if wait:\n ex_job.result()\n blobs_list = list(self.bucket.list_blobs(prefix=prefix))\n n_files = len(blobs_list) \n print(f'{n_files} file{\"s\" if n_files > 1 else \"\"} '\n f'exported to gcs with prefix {prefix}.')\n \n else:\n n_files = len(blobs_list) \n print(f'{n_files} file{\"s\" if n_files > 1 else \"\"} '\n f'already exist{\"s\" if n_files == 1 else \"\"} in gcs with '\n f'prefix {prefix}.')\n \n return prefix\n\n def get_table_gcs(self, prefix):\n \"\"\"Downloads all files at prefix if they don't exist locally.\n Returns list of file paths.\n \"\"\"\n \n file_paths = list(Path().glob(prefix))\n if not file_paths:\n blobs_list = list(self.bucket.list_blobs(prefix=prefix))\n n_files = len(blobs_list)\n print(f'Downloading {n_files} file{\"s\" if n_files > 1 else \"\"} '\n f'from gcs for table {prefix}...')\n\n for b in tqdm(blobs_list, desc='Files Downloaded: '):\n print('Downloading', b.name, b.size)\n Path(b.name).parent.mkdir(parents=True, exist_ok=True)\n b.download_to_filename(b.name)\n \n else:\n n_files = len(list(file_paths[0].iterdir()))\n print(f'{n_files} file{\"s\" if n_files > 1 else \"\"} already '\n f'exist{\"s\" if n_files == 1 else \"\"} locally for '\n f'table{prefix}.')\n \n return list(list(Path().glob(prefix))[0].iterdir())\n\n def get_df_files(self, file_paths, dtypes):\n \"\"\" Creates data frame from list of local file paths.\n Returns dataframe.\n \"\"\"\n \n prefix = str(file_paths[0].parent)\n suffix = file_paths[0].suffix\n \n n_files = len(file_paths)\n print(f'Creating dataframe from {n_files} file{\"s\" if n_files > 1 else \"\"} for table {prefix}...')\n \n dfs = []\n if suffix == '.csv':\n for f in tqdm(file_paths, desc='Files Read: '):\n dfs.append(pd.read_csv(f, dtype=dtypes))\n else:\n for f in tqdm(file_paths, desc='Files Read: '):\n dfs.append(pd.read_json(f, dtype=dtypes, lines=True))\n \n df_train = pd.concat(dfs)\n \n print(f'Dataframe finished for train table at {prefix} with'\n f' {len(df_train.columns):,d} columns and '\n f'{len(df_train):,d} rows.')\n \n return df_train\n\n def get_df_query_gcs(self, query, dtypes, file_format='csv', wait=True):\n prefix = self.export_query_gcs(query, file_format, wait)\n file_paths = self.get_table_gcs(prefix)\n df = self.get_df_files(file_paths, dtypes)\n \n return df# ","execution_count":null,"outputs":[]},{"metadata":{"id":"dTP7zTkV0qhK","_kg_hide-input":true,"trusted":false},"cell_type":"code","source":"import sys\n\nclass Queries:\n def __init__(self, DATASET):\n self.DATASET = DATASET\n \n def select_rows(self, table_id='train', limit=100):\n return f\"\"\"\n SELECT *\n FROM {self.DATASET}.{table_id}\n LIMIT {limit}\n \"\"\", sys._getframe().f_code.co_name + '_'\n\n def update_missing_values(self, table_id='train', column_id=None, value=None):\n return f\"\"\"\n UPDATE {self.DATASET}.{table_id}\n SET {column_id} = {value}\n WHERE {column_id} is NULL;\n \"\"\", sys._getframe().f_code.co_name + '_'\n\n def update_task_container_id(self, table_id='train', column_id_orig='task_container_id_orig'):\n return f\"\"\"\n UPDATE {self.DATASET}.{table_id}\n SET {column_id_orig} = task_container_id\n WHERE true;\n\n UPDATE {self.DATASET}.{table_id} t\n SET task_container_id = target.calc\n FROM (\n SELECT row_id, DENSE_RANK()\n OVER (\n PARTITION BY user_id\n ORDER BY timestamp\n ) - 1 calc\n FROM {self.DATASET}.{table_id}\n ) target\n WHERE target.row_id = t.row_id\n \"\"\", sys._getframe().f_code.co_name + '_'\n\n def create_train_sample(self, suffix='sample', user_id_max=50000):\n return f\"\"\"\n CREATE OR REPLACE TABLE {self.DATASET}.train_{suffix} AS\n SELECT *\n FROM {self.DATASET}.train\n WHERE user_id <= {user_id_max}\n ORDER BY user_id, task_container_id, row_id\n \"\"\", sys._getframe().f_code.co_name + '_'\n\n def select_train(self, columns=['*'], user_id_max=50000,\n excl_lectures=False, table_id='train'):\n \n where_condition = f'user_id <= {user_id_max}' if user_id_max else 'true'\n where_condition = (where_condition + ' AND content_type_id = 0'\n if excl_lectures else where_condition)\n \n return f\"\"\"\n SELECT {(', ').join(columns)}\n FROM {self.DATASET}.{table_id} t\n LEFT JOIN {self.DATASET}.questions q\n ON t.content_id = q.question_id\n WHERE {where_condition}\n ORDER BY user_id, task_container_id, row_id\n \"\"\", sys._getframe().f_code.co_name + '_'\n \n def update_answered_incorrectly(self, table_id='train'):\n \"\"\"Sets annswered_incorrectly to inverse of answered_correctly for questions.\n Sets answered_correctly to 0 for lectures so window totals for correct and\n incorrect are caculated correctly, including lectures.\n \"\"\"\n \n return f\"\"\"\n UPDATE {self.DATASET}.{table_id}\n SET answered_incorrectly = 0\n WHERE true;\n\n UPDATE {self.DATASET}.{table_id}\n SET answered_incorrectly = 1 - answered_correctly\n WHERE content_type_id = 0;\n\n UPDATE {self.DATASET}.{table_id}\n SET answered_correctly = 0\n WHERE content_type_id = 1;\n \"\"\", sys._getframe().f_code.co_name + '_'\n\n\n def update_questions_tag__0(self):\n return f\"\"\"\n UPDATE data.questions\n SET tag__0 = tags[OFFSET(0)]\n WHERE true;\n \"\"\", sys._getframe().f_code.co_name + '_' \n \n def update_train_window_containers(self, table_id='train'):\n return f\"\"\" \n UPDATE {self.DATASET}.{table_id} t\n SET answered_correctly_cumsum = IFNULL(calc.answered_correctly_cumsum, 0),\n answered_incorrectly_cumsum = IFNULL(calc.answered_incorrectly_cumsum, 0),\n lectures_cumcount = IFNULL(calc.lectures_cumcount, 0),\n prior_question_elapsed_time_rollavg = IFNULL(calc.prior_question_elapsed_time_rollavg, 0),\n answered_correctly_content_id_cumsum = IFNULL(calc.answered_correctly_content_id_cumsum, 0),\n answered_incorrectly_content_id_cumsum = IFNULL(calc.answered_incorrectly_content_id_cumsum, 0)\n FROM (\n SELECT row_id,\n SUM(answered_correctly) OVER (b) answered_correctly_cumsum,\n SUM(answered_incorrectly) OVER (b) answered_incorrectly_cumsum,\n SUM(content_type_id) OVER (b) lectures_cumcount,\n AVG(prior_question_elapsed_time) OVER (c) prior_question_elapsed_time_rollavg,\n SUM(answered_correctly) OVER (e) answered_correctly_content_id_cumsum,\n SUM(answered_incorrectly) OVER (e) answered_incorrectly_content_id_cumsum\n FROM {self.DATASET}.{table_id}\n WINDOW\n a AS (PARTITION BY user_id ORDER BY task_container_id),\n b AS (a RANGE BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING),\n c AS (a RANGE BETWEEN 3 PRECEDING AND 0 PRECEDING),\n d AS (PARTITION BY user_id, content_id ORDER BY task_container_id),\n e AS (d RANGE BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING)\n ORDER BY user_id, task_container_id, row_id\n ) calc\n WHERE calc.row_id = t.row_id\n \"\"\", sys._getframe().f_code.co_name + '_'\n\n def update_train_window_rows(self, table_id='train', window=10):\n \"\"\"Calculates aggregate over window number of rows with task_container_id\n less than task_container_id of current row.\n \"\"\"\n\n return f\"\"\" \n UPDATE {self.DATASET}.{table_id} u\n SET answered_correctly_rollsum = IFNULL(calc.answered_correctly_rollsum, 0),\n answered_incorrectly_rollsum = IFNULL(calc.answered_incorrectly_rollsum, 0)\n FROM (\n SELECT t.row_id,\n COUNT(j2.row_id) row_id_rollcount,\n SUM(j2.answered_correctly) answered_correctly_rollsum,\n SUM(j2.answered_incorrectly) answered_incorrectly_rollsum,\n FROM {self.DATASET}.{table_id} t\n JOIN (\n SELECT user_id, task_container_id, MIN(row_id) min_row\n FROM {self.DATASET}.{table_id}\n GROUP BY user_id, task_container_id\n ) j ON (j.user_id = t.user_id AND j.task_container_id = t.task_container_id)\n LEFT JOIN {self.DATASET}.{table_id} j2 ON (\n j2.user_id = t.user_id\n AND j2.task_container_id < t.task_container_id\n AND j2.row_id >= (j.min_row - {window + 1})\n )\n GROUP BY t.user_id, t.task_container_id, t.row_id\n ) calc\n WHERE\n calc.row_id = u.row_id\n \"\"\", sys._getframe().f_code.co_name + '_'\n\n\n def update_answered_correctly_cumsum_upto(self, table_id='train'): \n return f\"\"\" \n UPDATE {self.DATASET}.{table_id} t\n SET answered_correctly_cumsum_upto = IF(row_number < 11, r.answered_correctly_cumsum, m.ac_max)\n FROM (\n SELECT user_id, row_id, answered_correctly_cumsum,\n ROW_NUMBER() OVER(W) row_number,\n FROM {self.DATASET}.{table_id}\n WHERE content_type_id = 0\n WINDOW\n w AS (PARTITION BY user_id ORDER BY row_id)\n ) r\n JOIN (\n SELECT user_id, MAX(answered_correctly_cumsum) ac_max\n FROM (\n SELECT user_id, row_id, answered_correctly_cumsum,\n ROW_NUMBER() OVER(W) row_number,\n FROM {self.DATASET}.{table_id}\n WINDOW\n w AS (PARTITION BY user_id ORDER BY row_id)\n )\n WHERE row_number < 11\n GROUP BY user_id\n ) m\n ON (m.user_id = r.user_id)\n WHERE r.row_id = t.row_id\n \"\"\", sys._getframe().f_code.co_name + '_'\n\n def update_correct_cumsum_pct(self, column_id_correct=None,\n column_id_incorrect=None,\n update_column_id=None, table_id='train'):\n return f\"\"\"\n CREATE TEMP FUNCTION calcCorrectPct(c INT64, ic INT64) AS (\n CAST(SAFE_DIVIDE(c, (c + ic)) * 100 AS INT64)\n );\n\n UPDATE {self.DATASET}.{table_id}\n SET {update_column_id} =\n calcCorrectPct({column_id_correct}, {column_id_incorrect})\n WHERE true;\n \n UPDATE {self.DATASET}.{table_id}\n SET {update_column_id} = 0\n WHERE {update_column_id} IS NULL;\n \"\"\", sys._getframe().f_code.co_name + '_'\n\n def update_question_correct_pct(self, column_id):\n return f\"\"\" \n CREATE TEMP FUNCTION calcCorrectPct(c INT64, ic INT64) AS (\n CAST(SAFE_DIVIDE(c, (c + ic)) * 100 AS INT64)\n );\n\n UPDATE {self.DATASET}.questions q\n SET q.{column_id}_correct_pct = calcCorrectPct(c.c, c.ic)\n FROM (\n SELECT cq.{column_id}, SUM(answered_correctly) c, SUM(answered_incorrectly) ic\n FROM {self.DATASET}.train t\n JOIN {self.DATASET}.questions cq\n ON t.content_id = cq.question_id\n WHERE t.content_type_id = 0\n GROUP BY cq.{column_id}\n ) c\n WHERE q.{column_id} = c.{column_id}\n \"\"\", sys._getframe().f_code.co_name + '_'\n\n def select_user_id_rows(self, table_id='train', rows=30000):\n return f\"\"\" \n SELECT user_id\n FROM {self.DATASET}.{table_id}\n WHERE row_id = {rows}\n \"\"\", sys._getframe().f_code.co_name + '_'\n \n def select_user_final_state(self, table_id='train'):\n return f\"\"\" \n CREATE TEMP FUNCTION calcCorrectPct(c INT64, ic INT64) AS (\n IFNULL(CAST(SAFE_DIVIDE(c, (c + ic)) * 100 AS INT64), 0)\n );\n \n SELECT *, calcCorrectPct(answered_correctly_cumsum, answered_incorrectly_cumsum) answered_correctly_cumsum_pct,\n calcCorrectPct(answered_correctly_rollsum, answered_incorrectly_rollsum) answered_correctly_rollsum_pct\n FROM (\n SELECT row_id, user_id, answered_correctly_cumsum_upto, content_type_id,\n SUM(answered_correctly) OVER (b) answered_correctly_cumsum,\n SUM(answered_incorrectly) OVER (b) answered_incorrectly_cumsum,\n SUM(answered_correctly) OVER (d) answered_correctly_rollsum,\n SUM(answered_incorrectly) OVER (d) answered_incorrectly_rollsum,\n SUM(content_type_id) OVER (b) lectures_cumcount,\n AVG(prior_question_elapsed_time) OVER (c) prior_question_elapsed_time_rollavg,\n ROW_NUMBER() OVER(y) row_no_desc,\n SUM(answered_correctly + answered_incorrectly) OVER (d) answer_row_id_rollcount,\n SUM(answered_correctly + answered_incorrectly) OVER (c) time_row_id_rollcount,\n SUM(answered_correctly + answered_incorrectly) OVER (a) question_row_id_rollcount,\n FROM {self.DATASET}.{table_id}\n WINDOW\n x AS (PARTITION BY user_id),\n y AS (x ORDER BY task_container_id DESC, row_id DESC),\n a AS (x ORDER BY task_container_id),\n b AS (a ROWS BETWEEN UNBOUNDED PRECEDING AND 0 PRECEDING),\n c AS (a RANGE BETWEEN 3 PRECEDING AND 0 PRECEDING),\n d AS (a ROWS BETWEEN 9 PRECEDING AND 0 PRECEDING)\n )\n WHERE row_no_desc = 1 AND content_type_id = 0\n ORDER BY user_id\n \"\"\", sys._getframe().f_code.co_name + '_'\n\n def select_user_content_final_state(self, table_id='train'):\n return f\"\"\" \n SELECT user_id, content_id, SUM(answered_correctly) answered_correctly,\n SUM(answered_incorrectly) answered_incorrectly,\n FROM {self.DATASET}.{table_id}\n WHERE content_type_id = 0\n GROUP BY user_id, content_id\n ORDER BY user_id, content_id\n \"\"\", sys._getframe().f_code.co_name + '_'\n # ","execution_count":null,"outputs":[]},{"metadata":{"id":"FWx9otMx1O2Y","_kg_hide-input":true,"trusted":false},"cell_type":"code","source":"import os\nimport subprocess\n\ndef porc(c):\n print_output(run_command(c))\n\ndef print_output(output):\n \"\"\"Prints output from string.\"\"\"\n for l in output.split('\\n'):\n print(l)\n\ndef run_command(command):\n \"\"\"Runs command line command as a subprocess returning output as string.\"\"\"\n STDOUT = subprocess.PIPE\n process = subprocess.run(command, shell=True, check=False,\n stdout=STDOUT, stderr=STDOUT, universal_newlines=True)\n \n output = process.stdout if process.stdout else process.stderr\n \n return output\n\ndef get_v_tuple(v):\n return tuple([int(s) for s in v.split('.')])\n\ndef check_package(p, v):\n output = run_command(f'pip freeze | grep {p}==')\n if output == '':\n porc(f'pip install -q {p}')\n elif get_v_tuple(output[output.find('==')+2:]) < get_v_tuple(v):\n porc(f'pip install -q -U {p}')\n else:\n print_output(output)\n\ndef check_packages(packages):\n for p, v in packages.items():\n check_package(p, v)\n\nclass Git:\n def __init__(self, repo, username, password, email, base_path):\n self.repo = repo\n self.username = username\n self.password = password\n self.email = email\n self.repo_path = base_path/repo\n self.cred_repo = (\n f'https://{self.username}:{self.password}'\n f'@github.com/{self.username}/{self.repo}.git'\n )\n self.config()\n \n def config(self):\n commands = []\n commands.append(f'git config --global user.email {self.email}')\n commands.append(f'git config --global user.name {self.username}')\n for cmd in commands:\n porc(cmd)\n print('Git global user.name and user.email set.')\n \n def clone(self, latest=False):\n cwd = os.getcwd()\n os.chdir(self.base_path)\n\n if latest:\n cred_repo = f'--depth 1 {cred_repo}'\n\n commands = []\n commands.append(f'git clone {cred_repo}')\n for cmd in commands:\n porc(cmd)\n\n os.chdir(cwd)\n\n def commit(self, message='made some changes'):\n cwd = os.getcwd()\n os.chdir(self.repo_path)\n porc('git add -A')\n porc(f'git commit -m \"{message}\"')\n os.chdir(cwd)\n\n def command(self, command):\n cwd = os.getcwd()\n os.chdir(self.repo_path)\n porc(f'git {command}')\n os.chdir(cwd)\n\n def status(self):\n self.command('status')\n\n def push(self, branch='master'):\n self.command(f'push origin {branch}')\n\n def set_remote(self):\n self.command(f'remote set-url origin {self.cred_repo}')# ","execution_count":null,"outputs":[]},{"metadata":{"id":"r4nP2GeV09FG","_kg_hide-input":true,"_kg_hide-output":true,"trusted":false},"cell_type":"code","source":"{\"COMET_API_KEY\": \"\", \"COMET_PROJECT_NAME\": \"\", \"COMET_WORKSPACE\": \"\", \"EMAIL\": \"\", \"GIT_PASSWORD\": \"\", \"GIT_USERNAME\": \"\", \"KAGGLE_USERNAME\": \"\", \"KAGGLE_KEY\": \"\"}# ","execution_count":null,"outputs":[]},{"metadata":{"id":"WO6DVIxuQydR","trusted":false},"cell_type":"code","source":"Q = Queries(DATASET)\nbqh = BQHelper(bucket, DATASET, bq_client)","execution_count":null,"outputs":[]},{"metadata":{"id":"cs6bXr35juQM"},"cell_type":"markdown","source":"## Create BigQuery Dataset"},{"metadata":{"trusted":true,"id":"MVloHU3wjuQQ"},"cell_type":"code","source":"if False:\n delete_contents=False\n bq_client.delete_dataset(DATASET, delete_contents=delete_contents)\n print(f'Dataset {dataset.dataset_id} deleted from project {dataset.project}.')","execution_count":null,"outputs":[]},{"metadata":{"trusted":true,"id":"y2-E_hTejuQO"},"cell_type":"code","source":"try:\n dataset = bq_client.get_dataset(dataset.dataset_id)\n print(f'Dataset {dataset.dataset_id} already exists '\n f'in location {dataset.location} in project {dataset.project}.')\nexcept:\n dataset = bq_client.create_dataset(dataset)\n print(f'Dataset {dataset.dataset_id} created '\n f'in location {dataset.location} in project {dataset.project}.')","execution_count":null,"outputs":[]},{"metadata":{"id":"HItb6CWGjuQS"},"cell_type":"markdown","source":"## Load Tables"},{"metadata":{"id":"BjTuu8CJjuQT"},"cell_type":"markdown","source":"### Dataframe dtypes"},{"metadata":{"trusted":true,"id":"8dxMzeCsjuQT","_kg_hide-input":true},"cell_type":"code","source":"# \ndtypes_orig = {\n 'lectures': {\n 'lecture_id': 'uint16',\n 'tag': 'uint8',\n 'part': 'uint8',\n 'type_of': 'str',\n },\n 'questions': {\n 'question_id': 'uint16',\n 'bundle_id': 'uint16',\n 'correct_answer': 'uint8',\n 'part': 'uint8',\n 'tags': 'str',\n \n },\n 'train': {\n 'row_id': 'int64',\n 'timestamp': 'int64',\n 'user_id': 'int32',\n 'content_id': 'int16',\n 'content_type_id': 'int8',\n 'task_container_id': 'int16',\n 'user_answer': 'int8',\n 'answered_correctly': 'int8',\n 'prior_question_elapsed_time': 'float32', \n 'prior_question_had_explanation': 'bool'\n }\n \n}\n\ndtypes_new = {\n 'lectures': {},\n 'questions': {\n 'tag__0': 'uint8',\n 'part_correct_pct': 'uint8',\n 'tag__0_correct_pct': 'uint8',\n 'question_id_correct_pct': 'uint8'\n },\n 'train': {\n 'task_container_id_orig': 'int16',\n 'answered_correctly_cumsum': 'int16',\n 'answered_correctly_rollsum': 'int8',\n 'answered_incorrectly': 'int8',\n 'answered_incorrectly_cumsum': 'int16',\n 'answered_incorrectly_rollsum': 'int8',\n 'answered_correctly_cumsum_pct': 'int8',\n 'answered_correctly_rollsum_pct': 'int8',\n 'answered_correctly_content_id_cumsum': 'int16',\n 'answered_incorrectly_content_id_cumsum': 'int16',\n 'answered_correctly_content_id_cumsum_pct': 'int16',\n 'answered_correctly_cumsum_upto': 'int8',\n 'prior_question_elapsed_time_rollavg': 'float32',\n 'lectures_cumcount': 'int16',\n }\n}\n\none_hot_tags = False\nif one_hot_tags:\n for tag in range(189):\n for table_id in ['questions']:\n dtypes_new[table_id][f'tag_{tag:03d}'] = 'uint8'\n\ndtypes = {}\nfor table_id in dtypes_orig:\n dtypes[table_id] = {**dtypes_orig[table_id], **dtypes_new[table_id]}\n\ndtypes = {\n **dtypes['lectures'],\n **dtypes['questions'],\n **dtypes['train']\n}","execution_count":null,"outputs":[]},{"metadata":{"id":"QhPzQmvUjuQV"},"cell_type":"markdown","source":"### BigQuery Table Schemas"},{"metadata":{"trusted":true,"id":"WdW0igS4juQW","_kg_hide-input":true},"cell_type":"code","source":"# \ntype_map = {\n 'int64': 'INTEGER',\n 'int32': 'INTEGER',\n 'int16': 'INTEGER',\n 'int8': 'INTEGER',\n 'uint8': 'INTEGER',\n 'uint16': 'INTEGER',\n 'str': 'STRING',\n 'bool': 'BOOL',\n 'float32': 'FLOAT'\n}\n\nschemas_orig = {table: [SchemaField(f, type_map[t]) for f, t in\n fields.items()] for table, fields in dtypes_orig.items()}\nschemas_orig['questions'][-1] = SchemaField('tags', 'INTEGER', 'REPEATED')\n\nschemas = {}\nfor table_id, fields in dtypes_new.items():\n new_fields = [SchemaField(f, type_map[t]) for\n f, t in fields.items()]\n schemas[table_id] = schemas_orig[table_id] + new_fields","execution_count":null,"outputs":[]},{"metadata":{"id":"GOpOrtr9juQY"},"cell_type":"markdown","source":"### Load Tables"},{"metadata":{"trusted":true,"id":"saboKI_rjuQc","_kg_hide-input":true},"cell_type":"code","source":"# \n# Load questions from local json file - can't load tags as array from csv.\n\nif False:\n bqh.del_table('questions')\n \n df_questions = pd.read_csv(f'gs://{BUCKET}/questions.csv')\n df_questions.tags = df_questions.tags.fillna('188')\n df_questions.tags = df_questions.tags.str.split()\n \n if one_hot_tags:\n mlb = MultiLabelBinarizer()\n one_hots = (mlb.fit_transform(df_questions.tags\n .apply(lambda l: [f'tag_{int(t):03d}' for t in l])))\n df_one_hots = pd.DataFrame(one_hots, columns = mlb.classes_)\n df_questions = pd.concat([df_questions, df_one_hots], axis=1)\n \n df_questions.to_json('questions.json', orient=\"records\", lines=True)\n lj = bqh.load_json_file('questions', schemas).result()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true,"id":"XXwb4YkWjuQe","_kg_hide-input":true},"cell_type":"code","source":"# \nif False:\n for table_id in ['lectures', 'train']:\n bqh.del_table(table_id)\n lj = bqh.load_csv_uri(table_id, schemas_orig).result()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true,"id":"CHkK8JhQjuQg","_kg_hide-input":true},"cell_type":"code","source":"# \ndf_jobs = bqh.get_df_jobs()\ndf_jobs","execution_count":null,"outputs":[]},{"metadata":{"trusted":true,"id":"5cGVHAtMjuQi","_kg_hide-input":true},"cell_type":"code","source":"# \ndf_table_list = bqh.get_df_table_list()\ndf_table_list","execution_count":null,"outputs":[]},{"metadata":{"id":"sgTZkanIjuQk"},"cell_type":"markdown","source":"### Update Table Schemas"},{"metadata":{"trusted":true,"id":"_t7T4NftjuQk","_kg_hide-input":true},"cell_type":"code","source":"# \nif False:\n for table_id, schema in schemas.items():\n table = bqh.get_table(table_id)\n table.schema = schema\n table = bq_client.update_table(table, ['schema'])","execution_count":null,"outputs":[]},{"metadata":{"id":"Ouoo1oQCjuQm"},"cell_type":"markdown","source":"## Engineer Features"},{"metadata":{"id":"CU8m9kx9juQn"},"cell_type":"markdown","source":"A good workflow here is:\n* Create a sample of the train table.\n* Use the BigQuery query editor user interface to get the SQL for a new feature worked out as a selection from the `train_sample` table. The user interface there has tab completion, syntax checking and displays results, which makes creating and debugging queries a snap.\n * [BigQuery Console](https://console.cloud.google.com/bigquery?project=riiid-caleb) (Update project query string for your project.)\n * [BigQuery Query syntax in Standard SQL](https://cloud.google.com/bigquery/docs/reference/standard-sql/query-syntax) is your friend.\n* Optional: create a local dataframe, using the export functions below, to confirm that it is working the right way.\n* Add a column to the appropriate table by adding a value to `dtypes_new`\n* Update the schema for the table in BigQuery by running the Update Table Schemas cell above\n* Recreate the `train_sample` table by running the cell below.\n* Use the BigQuery query editor user interface add the logic to update the new column.\n* Optional: create a local dataframe, using the export functions below, to confirm that the update is working the right way.\n* Copy the SQL to a new method in the `Queries` class above\n* Add the query to the appropriate `run_transformations` function above\n* Run transformations on `train_sample` table\n* Inspect `train_sample` table in BigQuery to confirm everything is working correctly\n* Optional: load load local dataframe using `get_df_query` function for further inspection\n* Run transformations on `train` table\n* Inspect `train` table in BigQuery to confirm everything is working correctly\n* Optional: load local dataframe using `get_df_query` function for further inspection"},{"metadata":{"id":"CfKY_WsvjuQp"},"cell_type":"markdown","source":"### Perform Transformations"},{"metadata":{"id":"ZOnBP6sfjuQp"},"cell_type":"markdown","source":"#### Train Table\n* Add question columns\n * Adding question part and the first associated tag. (There wasn't any official information regarding the order of the tags as recorded for each question, but they did not appear to be sorted so it seems possible the order in which they are recorded is significant.)\n* Update task_container_id to increase monotonically with timestamp\n * There were some `task_conatiner_id`s that were out of order with respect to timestamp. They needed to be be ordered correctly so that cumulative and rolling sums partitioned by `task_container_id` would be include only interactions with earlier `timestamps`. Even though all interactions with the same `task_container_id` have the same `timestamp`, partioning by `timestamp` is much slower (because the range of values is so much wider?).\n* Calc answered_incorrectly\n * `answered_correctly` for lectures was recorded as -1 and needed to be set to 0 to calculate cumulative and rolling sums correctly including lectures. As a consequence, `answered_incorrectly` could be calculated as the inverse of `answered_correctly`.\n* Calc cumsum for `answered_correctly` and `answered_incorrectly` by `user_id` and by `user_id` and `content_id` and rolling avg for `prior_question_elapsed_time` by user \n * This is done so that the totals are as of the preceding `task_container_id`\n* Calculate rolling sum for `answered_correctly` and `answered_incorrectly` by `user_id`\n * Includes the 10 rows preceding the current `task_container_id`\n * I couldn't figure out how to get this done with the standard window functionality since I wanted a set number of rows preceding the current task container (as opposed to just the current row), so it joins on `user_id` with a `task_container_id` less than the current one, which takes a while to complete.\n* Calculate answered correctly percentages for `answered_correctly_cumsum`, `answered_correctly_rollsum` and `answered_correctly_content_id_cumsum_pct`\n\n#### Questions Table\n* Calculate percent answered correctly for `question_id`, `part` and `tag__0` "},{"metadata":{"id":"ndY56UJdhrIU","_kg_hide-input":true,"trusted":false},"cell_type":"code","source":"# \ncumsum_pct_specs = [\n dict(column_id_correct='answered_correctly_cumsum',\n column_id_incorrect='answered_incorrectly_cumsum',\n update_column_id='answered_correctly_cumsum_pct'),\n \n dict(column_id_correct='answered_correctly_rollsum',\n column_id_incorrect='answered_incorrectly_rollsum',\n update_column_id='answered_correctly_rollsum_pct'),\n \n dict(column_id_correct='answered_correctly_content_id_cumsum',\n column_id_incorrect='answered_incorrectly_content_id_cumsum',\n update_column_id='answered_correctly_content_id_cumsum_pct'), \n]\n\ndef run_update_correct_cumsum_pct(spec):\n query, job_id_prefix = Q.update_correct_cumsum_pct(**spec)\n job_id_prefix = f'{job_id_prefix}{spec[\"update_column_id\"]}_'\n bqh.run_query(query=query, job_id_prefix=job_id_prefix, wait=True)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true,"id":"r2LZdBETjuQs","_kg_hide-input":true},"cell_type":"code","source":"# \ndef run_train_transforms(table_id=None):\n # Run serially to avoid update conflicts\n \n train_queries = [\n Q.update_task_container_id(table_id=table_id),\n Q.update_answered_incorrectly(table_id=table_id),\n Q.update_missing_values(table_id=table_id,\n column_id='prior_question_had_explanation',\n value='false'),\n Q.update_missing_values(table_id=table_id,\n column_id='prior_question_elapsed_time',\n value='0'),\n Q.update_train_window_containers(table_id=table_id),\n Q.update_train_window_rows(table_id=table_id, window=10),\n Q.update_answered_correctly_cumsum_upto(table_id=table_id)\n ]\n \n _ = [bqh.run_query(*q, wait=True) for q in train_queries]\n\n _ = [spec.update(table_id=table_id) for spec in cumsum_pct_specs]\n _ = list(map(run_update_correct_cumsum_pct, cumsum_pct_specs))","execution_count":null,"outputs":[]},{"metadata":{"trusted":true,"id":"8Ay24JcgjuQv","_kg_hide-input":true},"cell_type":"code","source":"# \ndef run_questions_transforms():\n \"\"\"These have to be run after the transforms are run on the full\n train table.\n \"\"\"\n \n questions_queries = [Q.update_questions_tag__0()]\n for column_id in ['question_id', 'part', 'tag__0']:\n questions_queries.append(Q.update_question_correct_pct(column_id))\n \n _ = [bqh.run_query(*q, wait=True).result() for q in questions_queries]","execution_count":null,"outputs":[]},{"metadata":{"trusted":true,"id":"tuh_k3VfjuQx","_kg_hide-input":true},"cell_type":"code","source":"# \nif False:\n run_train_transforms('train')\n run_questions_transforms()","execution_count":null,"outputs":[]},{"metadata":{"id":"YTG_h2a2juQy"},"cell_type":"markdown","source":"### Check Output"},{"metadata":{"trusted":true,"id":"12-1u5TdjuQ1"},"cell_type":"code","source":"query = Q.select_train(table_id='train', excl_lectures=True)\ndf_query = bqh.get_df_query(query, dtypes=dtypes)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true,"id":"ub-U0nPgjuQ3","_kg_hide-input":true},"cell_type":"code","source":"# \ncols = [\n 'row_id',\n 'task_container_id_orig',\n 'timestamp',\n 'content_type_id',\n 'user_id',\n 'task_container_id',\n 'part',\n 'tag__0',\n 'answered_correctly',\n 'answered_incorrectly',\n 'answered_correctly_cumsum',\n 'answered_incorrectly_cumsum',\n 'answered_correctly_content_id_cumsum',\n 'answered_correctly_rollsum',\n 'answered_incorrectly_rollsum',\n 'answered_incorrectly_content_id_cumsum',\n 'part_correct_pct',\n 'tag__0_correct_pct',\n 'question_id_correct_pct',\n 'prior_question_elapsed_time',\n 'prior_question_elapsed_time_rollavg',\n 'prior_question_had_explanation',\n 'lectures_cumcount',\n 'answered_correctly_cumsum_upto'\n]\n\ndf_user = df_query[cols].copy()\ndf_user.timestamp = df_user.timestamp / (1000*60*60)\n\ndf_user.loc[df_user.user_id == 44331].head(20)","execution_count":null,"outputs":[]},{"metadata":{"id":"zO6GM_CLjuQ5"},"cell_type":"markdown","source":"### Visually Inspect Features"},{"metadata":{"id":"jpokxssbjuQ6"},"cell_type":"markdown","source":"The charts below can also be used to visually inspect whether the transformations have been performed correctly."},{"metadata":{"trusted":true,"id":"0c-xlk2TjuQ6","_kg_hide-input":true},"cell_type":"code","source":"# \ngroups = {\n 'cum': {\n 'columns': {\n 'task_container_id': 0,\n 'answered_correctly_cumsum': 2,\n 'answered_incorrectly_cumsum': 1\n },\n 'xaxis': 'elapsed_hours'\n },\n 'roll': {\n 'columns': {\n 'answered_correctly_rollsum': 2,\n 'answered_correctly': 7,\n 'answered_incorrectly_rollsum': 1,\n 'answered_incorrectly': 8,\n 'part': 9\n },\n 'xaxis': 'row_id'\n }, \n 'correct_pct': {\n 'columns': {\n 'question_id_correct_pct': 0,\n 'part_correct_pct': 5,\n 'tag__0_correct_pct': 6\n },\n 'xaxis': 'row_id'\n }, \n 'prior_question_elapsed_time': {\n 'columns': {\n 'prior_question_elapsed_time': 0,\n },\n 'xaxis': 'row_id'\n }, \n 'prior_question_had_explanation': {\n 'columns': {\n 'prior_question_had_explanation': 0,\n },\n 'xaxis': 'row_id'\n }\n}\n\ndef plot_user_learning(user_id=None, group=None, suffix=None):\n theme = px.colors.qualitative.Plotly\n columns = list(group['columns'].keys())\n colors = [theme[c] for c in group['columns'].values()]\n\n df_query['elapsed_hours'] = df_query.timestamp / (1000*60*60)\n\n df = (df_query.loc[(df_user.user_id == user_id) &\n (df_user.content_type_id == 0)])\n\n # labels = {'value': 'answer count'}\n\n fig = df.plot(x=group['xaxis'], y=columns, color_discrete_sequence=colors,\n title=f'Learning Progress - user_id = {user_id} - {suffix}')\n fig.data\n\n return fig\n\nuser_id_random = np.random.choice(df_query.user_id.unique(), (1,))[0]\nuse_random = False\nuser_id = user_id_random if use_random else 5382\n\nfor k, v in groups.items():\n fig = plot_user_learning(user_id, v, k)\n fig.show()","execution_count":null,"outputs":[]},{"metadata":{"id":"F_B3C-9QjuQ9"},"cell_type":"markdown","source":"### Create Sample of Train Table for R&D"},{"metadata":{"trusted":true,"id":"tuNFCT1OjuQ9","_kg_hide-input":true},"cell_type":"code","source":"# \nif False:\n bqh.run_query(*Q.create_train_sample(), wait=True)\n q = Q.select_train(excl_lectures=True, table_id='train_sample')\n df_sample = bqh.get_df_query(q)","execution_count":null,"outputs":[]},{"metadata":{"id":"wbaxOZFVjuRA"},"cell_type":"markdown","source":"## Create Local Training Dataframe"},{"metadata":{"id":"7GBDKOZ2juRA"},"cell_type":"markdown","source":"With feature engineering being performed in BigQuery, data has to be exported to train models locally. The [Python Client for Google BigQuery](https://googleapis.dev/python/bigquery/latest/index.html) [to_dataframe()](https://googleapis.dev/python/bigquery/latest/generated/google.cloud.bigquery.job.QueryJob.html?highlight=to_dataframe#google.cloud.bigquery.job.QueryJob.to_dataframe) makes it possible to create dataframes directly, but is prohibitively slow for large datasets. While it is not possible to export table directly to the local file system, it is possible to export to cloud storage and then download locally from there. This is reasonably efficient, taking a couple of minutes to run a query, export to cloud storage, download to the local file system and then read the files into a dataframe. The is another api, the [BigQuery Storage API](https://cloud.google.com/bigquery/docs/reference/storage), that a client can be created with that is really fast and works with the `to_dataframe` method, but unforunatley it isn't working with the current Kaggle kernel environment.\n\nThe functions below take advantage of the fact BigQuery stores queries in temporary tables so that preveiously requested queries can be retrieved without having to run them again. Similarly, the functions below name the exported files with the reference to the BigQuery temporary table, so that if a function is run to create a dataframe from a query for which the files already exist in cloud storage or locally, they won't be exported or downloaded again. "},{"metadata":{"id":"273_RZ1StVS0"},"cell_type":"markdown","source":"### Create DataFrame"},{"metadata":{"trusted":true,"id":"QUkacRjajuRG","_kg_hide-input":true},"cell_type":"code","source":"# \nfeatures = {\n 'answered_correctly': True,\n 'answered_correctly_content_id_cumsum': True,\n 'answered_correctly_content_id_cumsum_pct': True,\n 'answered_correctly_cumsum': True,\n 'answered_correctly_cumsum_upto': True,\n 'answered_correctly_cumsum_pct': True,\n 'answered_correctly_rollsum': True,\n 'answered_correctly_rollsum_pct': True,\n 'answered_incorrectly': True,\n 'answered_incorrectly_content_id_cumsum': True,\n 'answered_incorrectly_cumsum': True,\n 'answered_incorrectly_rollsum': True,\n 'bundle_id': False,\n 'content_id': True,\n 'content_type_id': True,\n 'correct_answer': False,\n 'lecture_id': False,\n 'lectures_cumcount': True,\n 'part': True,\n 'part_correct_pct': True,\n 'prior_question_elapsed_time': True,\n 'prior_question_elapsed_time_rollavg': True,\n 'prior_question_had_explanation': True,\n 'question_id': False,\n 'question_id_correct_pct': True,\n 'row_id': True,\n 'tag': False,\n 'tag__0': True,\n 'tag__0_correct_pct': True,\n 'tags': False,\n 'task_container_id': True,\n 'task_container_id_orig': False,\n 'timestamp': True,\n 'type_of': False,\n 'user_answer': False,\n 'user_id': True\n}\n\ntag_cols = [f'tag_{tag:03d}' for tag in range(189)] if one_hot_tags else []\n\ncolumns_export = [f for f, v in features.items() if v]\nif one_hot_tags:\n columns_export = columns_export + tag_cols","execution_count":null,"outputs":[]},{"metadata":{"id":"32Wt1DvjpnJZ","_kg_hide-input":true,"trusted":false},"cell_type":"code","source":"# \ndef get_features_widget(features_dict, columns_list):\n\n names = []\n widget_list = []\n for key, v in features_dict.items():\n widget_list.append(widgets.ToggleButton(value=v,\n description=key,\n layout={'width': '290px'},\n button_style='primary'))\n names.append(key)\n\n arg_dict = {names[i]: widget for i, widget in enumerate(widget_list)}\n\n layout = widgets.Layout(grid_template_columns=\"repeat(3, 300px)\")\n ui = widgets.GridBox(widget_list, layout=layout)\n\n def select_data(**kwargs):\n columns_list.clear()\n\n for key in kwargs:\n features_dict[key] = False\n if kwargs[key]:\n columns_list.append(key)\n features_dict[key] = True\n\n print(f'{len(columns_list)} columns selected')\n\n output = widgets.interactive_output(select_data, arg_dict)\n return ui, output","execution_count":null,"outputs":[]},{"metadata":{"id":"B6ddqQ4HBT3R","_kg_hide-input":true,"trusted":false},"cell_type":"code","source":"# \ncolumns_export = []\ndisplay(*get_features_widget(features, columns_export))","execution_count":null,"outputs":[]},{"metadata":{"trusted":true,"id":"7eS8EvZGjuRI","_kg_hide-input":true},"cell_type":"code","source":"# \n# get user_ids for rows in thousands. will be approximate, excludes lectures\n# and selects all records for user_ids less than specified.\n\nif False: \n r = Q.run_query(*Q.select_user_id_rows(rows=int(2e6))).result()\n user_id_max = list(r)[0].user_id\n print(user_id_max)\n \nuser_ids = {\n 10: 91216,\n 100: 2078569,\n 1000: 20949024,\n 2000: 42207371,\n 10000: 216747867,\n 30000: 643006676\n}","execution_count":null,"outputs":[]},{"metadata":{"trusted":true,"id":"EXRXQ86UjuRK","_kg_hide-output":true},"cell_type":"code","source":"# \nif True:\n query = Q.select_train(columns=columns_export, user_id_max=user_ids[10000],\n excl_lectures=True)\n df_train = bqh.get_df_query_gcs(query, dtypes=dtypes, file_format='json')","execution_count":null,"outputs":[]},{"metadata":{"id":"oMqolxVmjuRL"},"cell_type":"markdown","source":"## Train Model"},{"metadata":{"id":"Axb2oD77juRM"},"cell_type":"markdown","source":"### Create Train and Validation Splits"},{"metadata":{"id":"jNsk1_JsjuRN"},"cell_type":"markdown","source":"This is a first pass at a validation split to be able to have something to get the mechanics of evaluating the model up and running. It simply takes the last 20 `task_container_id`s for each user. The result is that all of the records in the validation set have `task_container_ids` greater than those in the training set for each user. There are also users in the validation set that are not present in the training set. However, a significant problem with this methodology is that the number of records per user in the validation set is much lower than it is in the training set."},{"metadata":{"trusted":true,"id":"E791NgaXjuRN","_kg_hide-input":true},"cell_type":"code","source":"# \n# get unique user_id-task_container_id combinations\ndf_user_task = df_train.groupby(['user_id',\n 'task_container_id'])[['user_id',\n 'task_container_id',\n 'row_id']].head(1)\n\n# get index of trailing number of unique user_id-task_container_id combinations\nindex_valid = (df_user_task.groupby('user_id').tail(20)\n .set_index(['user_id', 'task_container_id']).index)\n\n# use index to get ids of all rows in the chosen set of user_id-task_container\n# combinations\nrow_valid = (df_train.set_index(['user_id', 'task_container_id'])['row_id']\n .loc[index_valid].values)\n\ndf_train['valid'] = df_train.row_id.isin(row_valid)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true,"id":"dUhlNdidjuRP","_kg_hide-input":true},"cell_type":"code","source":"# \ntitle = 'Train and Validation Splits - Record Counts'\ndf_train.valid.value_counts().plot(kind='bar', title=title)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true,"id":"FNj-Z1RajuRQ","_kg_hide-input":true},"cell_type":"code","source":"# \n(df_train.groupby(['valid','user_id'])[['valid','user_id']].head(1)\n .reset_index().groupby('valid').count().user_id\n .plot(kind='bar', title='Count of Users by Split'))","execution_count":null,"outputs":[]},{"metadata":{"trusted":true,"id":"h98GAgwRjuRU","_kg_hide-input":true},"cell_type":"code","source":"# \ng_user_ct = (df_train[['valid', 'row_id', 'user_id']]\n .groupby(['valid', 'user_id']).count())\n\nbins = [0,10,20,50,100,250,500,1000,2500,5000,20000]\ng_user_ct['bin'] = pd.cut(g_user_ct.row_id, bins=bins, duplicates='drop')\ng_counts = (g_user_ct.reset_index()\n .groupby(['valid', 'bin'])['row_id'].count().reset_index())\n\npx.bar(x=g_counts.bin.apply(str), y=g_counts.row_id,\n facet_col=g_counts.valid.map({True: 'Validation', False: 'Train'}),\n title='Count of Users by Count of Interactions by Split',\n labels={'x': 'Count of Interactions',\n 'y': 'Count of Users',\n 'facet_col': 'Validation Split'})","execution_count":null,"outputs":[]},{"metadata":{"id":"8VM6n9FKyczN"},"cell_type":"markdown","source":"### Select Columns for Training"},{"metadata":{"id":"Kr1aADasnT-M","_kg_hide-input":true,"trusted":false},"cell_type":"code","source":"# \nfeatures_train = {\n 'answered_correctly': False,\n 'answered_correctly_content_id_cumsum': True,\n 'answered_correctly_content_id_cumsum_pct': False,\n 'answered_correctly_cumsum': True,\n 'answered_correctly_cumsum_upto': False,\n 'answered_correctly_cumsum_pct': True,\n 'answered_correctly_rollsum': False,\n 'answered_correctly_rollsum_pct': False,\n 'answered_incorrectly': False,\n 'answered_incorrectly_content_id_cumsum': True,\n 'answered_incorrectly_cumsum': True,\n 'answered_incorrectly_rollsum': False,\n 'bundle_id': False,\n 'content_id': False,\n 'content_type_id': False,\n 'correct_answer': False,\n 'lecture_id': False,\n 'lectures_cumcount': False,\n 'part': True,\n 'part_correct_pct': True,\n 'prior_question_elapsed_time': False,\n 'prior_question_elapsed_time_rollavg': False,\n 'prior_question_had_explanation': False,\n 'question_id': False,\n 'question_id_correct_pct': True,\n 'row_id': False,\n 'tag': False,\n 'tag__0': True,\n 'tag__0_correct_pct': True,\n 'tags': False,\n 'task_container_id': True,\n 'task_container_id_orig': False,\n 'timestamp': True,\n 'type_of': False,\n 'user_answer': False,\n 'user_id': False\n }\n\ncolumns_train = [f for f, v in features_train.items() if v] + tag_cols","execution_count":null,"outputs":[]},{"metadata":{"id":"MUSDoZoLEKP_","_kg_hide-input":true,"trusted":false},"cell_type":"code","source":"# \ncolumns_train = []\ndisplay(*get_features_widget(features_train, columns_train))","execution_count":null,"outputs":[]},{"metadata":{"id":"q-zrGOfDwjw9","_kg_hide-input":true,"trusted":false},"cell_type":"code","source":"# \ndef show_features():\n df_features = pd.DataFrame([features, features_train]).T.reset_index()\n df_features.columns = ['feature', 'export', 'train']\n df_features\n\n def highlight_true(s):\n return ['background-color: lightskyblue' if v else '' for v in s]\n return df_features.style.apply(highlight_true, subset=['export', 'train'])\nshow_features()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true,"id":"reut9PmcjuRV","_kg_hide-input":true},"cell_type":"code","source":"# \ny_train_col = ['answered_correctly']\n\nx_train_cols = columns_train\n\ntrain_matrix = xgb.DMatrix(data=df_train.loc[~df_train.valid][x_train_cols],\n label=df_train.loc[~df_train.valid][y_train_col])\n\nvalid_matrix = xgb.DMatrix(data=df_train.loc[df_train.valid][x_train_cols],\n label=df_train.loc[df_train.valid][y_train_col])","execution_count":null,"outputs":[]},{"metadata":{"id":"E5gUTzNtjuRX"},"cell_type":"markdown","source":"### Train Model"},{"metadata":{"trusted":true,"id":"t6AeR8lBjuRX","_kg_hide-output":true},"cell_type":"code","source":"# \nparams = {\n 'eta': 0.2,\n 'max_depth': 6,\n 'max_bin': 256,\n 'tree_method': 'gpu_hist',\n 'grow_policy': 'lossguide',\n 'sampling_method': 'gradient_based',\n 'objective': 'binary:logistic',\n 'eval_metric': ['error', 'logloss', 'auc']\n}\n\nif NOT_KAGGLE:\n experiment = Experiment()\n\nevals_result = {}\nmodel = xgb.train(params=params, dtrain=train_matrix, num_boost_round=300,\n evals=[(train_matrix, 'train'), (valid_matrix, 'valid')],\n evals_result=evals_result, early_stopping_rounds=10)\n\nif NOT_KAGGLE:\n experiment.end()","execution_count":null,"outputs":[]},{"metadata":{"id":"fe7JKQf0juRZ"},"cell_type":"markdown","source":"## Evaluate Model"},{"metadata":{"trusted":true,"id":"qPTKUFJMjuRZ","_kg_hide-input":true},"cell_type":"code","source":"# \ndef get_evals_df(evals_result):\n evals_list = []\n for k,v in evals_result.items():\n for j,u in v.items():\n evals_list.extend([{'epoch': i,\n 'split': k,\n 'metric': j,\n 'result': r} for i,r in enumerate(u)])\n \n df_evals = (pd.DataFrame(evals_list).set_index(['split', 'metric', 'epoch'])\n .unstack('metric'))\n df_evals.columns = df_evals.columns.get_level_values(1)\n df_evals.columns.name = None\n \n return df_evals.reset_index()\n\ndf_evals = get_evals_df(evals_result)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true,"id":"oa7h74fSjuRb","_kg_hide-input":true},"cell_type":"code","source":"# \ndf_evals.plot(x='epoch', y=['auc', 'logloss'],\n facet_col='split', title='Learning Curves')","execution_count":null,"outputs":[]},{"metadata":{"trusted":true,"id":"iIK9aXVpjuRd","_kg_hide-input":true},"cell_type":"code","source":"# \nimps = model.get_score(importance_type='gain').items()\ndf_imp = pd.DataFrame(imps, columns=['feature', 'importance'])\ndf_imp = df_imp.set_index('feature').sort_values('importance', ascending=False)\ndf_imp.plot(kind='bar', y='importance', title='Feature Importances - Gain')","execution_count":null,"outputs":[]},{"metadata":{"id":"t49TzqNPq5pg"},"cell_type":"markdown","source":"## Prepare Prediction Data"},{"metadata":{"id":"GXZ10jxVjuRf"},"cell_type":"markdown","source":"### Download Final Users State"},{"metadata":{"trusted":true,"id":"3eJlJt1FjuRg","_kg_hide-input":true,"_kg_hide-output":true},"cell_type":"code","source":"# \n# not using this currently, creating dataframe from users-content\n# table in submission notebook\nif False:\n query = Q.select_user_final_state(table_id='train')\n prefix = bqh.export_query_gcs(query, wait=True)\n file_paths = bqh.get_table_gcs(prefix)\n df_users = (bqh.get_df_files(file_paths, dtypes=dtypes)\n .reset_index(drop=True).set_index('user_id'))","execution_count":null,"outputs":[]},{"metadata":{"id":"Xok_9xlD7c0H"},"cell_type":"markdown","source":"### Download Final User-Content State"},{"metadata":{"id":"RUVdoI5V8FD1","_kg_hide-input":true,"_kg_hide-output":true,"trusted":false},"cell_type":"code","source":"# \nif False:\n query = Q.select_user_content_final_state(table_id='train')\n prefix = bqh.export_query_gcs(query, wait=True)\n file_paths = bqh.get_table_gcs(prefix)\n df_users_content = (bqh.get_df_files(file_paths, dtypes=dtypes)\n .sort_values(['user_id', 'content_id']))","execution_count":null,"outputs":[]},{"metadata":{"id":"tWvj0FYpjuRj"},"cell_type":"markdown","source":"### Download Questions Table"},{"metadata":{"trusted":true,"id":"mG3gRUwmjuRl","_kg_hide-input":true,"_kg_hide-output":true},"cell_type":"code","source":"# \nif False:\n # only 13k rows, so it downloaded directly from BigQuery\n df_questions = bqh.get_df_table('questions',\n max_results=None,\n dtypes=dtypes).sort_values('question_id')","execution_count":null,"outputs":[]},{"metadata":{"id":"_NH5WM3VjuRn"},"cell_type":"markdown","source":"### Update Submission Dataset"},{"metadata":{"trusted":true,"id":"x0B8sbkGjuRo","_kg_hide-input":true},"cell_type":"code","source":"# \nif False:\n Path(KAGGLE_SUBMIT_DATASET).mkdir(exist_ok=True)\n\n model.save_model(f'{KAGGLE_SUBMIT_DATASET}/model.xgb')\n\n with open(f'{KAGGLE_SUBMIT_DATASET}/columns.json', 'w') as cj:\n json.dump(columns_train, cj)\n \n df_files = {\n # 'df_users.pkl': df_users,\n 'df_users_content.pkl': df_users_content,\n 'df_questions.pkl': df_questions,\n }\n\n for file_path, df in df_files.items():\n df.to_pickle(f'{KAGGLE_SUBMIT_DATASET}/{file_path}')\n \n kaggle_id = f\"{os.getenv('KAGGLE_USERNAME')}/{KAGGLE_SUBMIT_DATASET}\"\n \n metadata = {\n \"licenses\": [{\"name\": \"CC0-1.0\"}],\n \"id\": kaggle_id,\n \"title\": KAGGLE_SUBMIT_DATASET\n }\n\n with open(f'{KAGGLE_SUBMIT_DATASET}/dataset-metadata.json', 'w') as f:\n json.dump(metadata, f)\n \n if kaggle_api.dataset_status(kaggle_id):\n kaggle_api.dataset_create_version(KAGGLE_SUBMIT_DATASET,\n version_notes='update dataset',\n delete_old_versions=True,\n dir_mode='tar',\n quiet=True\n )\n else:\n kaggle_api.dataset_create_new(KAGGLE_SUBMIT_DATASET,\n dir_mode='tar', quiet=True)","execution_count":null,"outputs":[]},{"metadata":{"id":"HN5k3_N-juRp"},"cell_type":"markdown","source":"## Submit From Kernel"},{"metadata":{"id":"BGXlGXmjjuRq"},"cell_type":"markdown","source":"* Go to [RIIID Submit](https://www.kaggle.com/calebeverett/riiid-submit), fork and update to reference your dataset."},{"metadata":{"id":"d-RK0etUwdw3"},"cell_type":"markdown","source":"## Push Kernel to Kaggle"},{"metadata":{"id":"XoJcDapFwhCT","_kg_hide-input":true,"trusted":false},"cell_type":"code","source":"# \nif NOT_KAGGLE:\n if True:\n \n code_file = 'riiid-2020.ipynb'\n with open(DRIVE/REPO/code_file, 'r') as nb:\n nb_json = json.load(nb) \n \n for i, cell in enumerate(nb_json['cells']):\n if cell['cell_type'] == 'code':\n \n # update show/hide code cells\n for h in ['input', 'output']:\n if cell['source'][0].find(f' 1:\n nb_json['cells'][i]['metadata'].update({f'_kg_hide-{h}': True})\n else:\n nb_json['cells'][i]['metadata'].pop(f'_kg_hide-{h}', None)\n\n # add modules as cells\n if len(cell['source']) == 1:\n groups = re.search(r'(?<=\\)', cell['source'][0])\n \n if groups:\n with open(DRIVE/REPO/groups.group(0), 'r') as m:\n nb_json['cells'][i]['source'] = m.readlines() + nb_json['cells'][i]['source'] \n\n\n if Path(code_file).exists():\n Path(code_file).unlink()\n \n with open(f'{code_file}', 'w') as f:\n json.dump(nb_json, f)\n\n data = {'id': 'calebeverett/riiid-bigquery-xgboost-end-to-end',\n 'title': 'RIIID: BigQuery-XGBoost End-to-End',\n 'code_file': code_file,\n 'language': 'python',\n 'kernel_type': 'notebook',\n 'is_private': 'false',\n 'enable_gpu': 'true',\n 'enable_internet': 'true',\n 'dataset_sources': [],\n 'competition_sources': ['riiid-test-answer-prediction'],\n 'kernel_sources': []}\n \n with open('kernel-metadata.json', 'w') as f:\n json.dump(data, f)\n\n kaggle_api.kernels_push('.')","execution_count":null,"outputs":[]}],"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"pygments_lexer":"ipython3","nbconvert_exporter":"python","version":"3.6.4","file_extension":".py","codemirror_mode":{"name":"ipython","version":3},"name":"python","mimetype":"text/x-python"}},"nbformat":4,"nbformat_minor":4}