{"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_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"# Parquet Data Regrouping\n\nThe time series data in `train_series.parquet` is huge. Reading the entire data into memory at once is not a good practice, limiting the number of features that can be extracted, and slowing down the data pipelines.\n\nThe particular dataset in this competition is segmented into different series_ids. Each series_id's data should be grouped together to have a robust cross validation. Otherwise data leaking from train to validation test can hurt the models' performances. How can we access the data better?\n\nThe answer is to regroup the data. Here I'll regroup the data by series_id and write a new parquet file. Working with this new file will be much faster.","metadata":{}},{"cell_type":"code","source":"%%time\nimport pyarrow as pa\nimport pyarrow.parquet as pq\nfrom tqdm import tqdm","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","execution":{"iopub.status.busy":"2023-09-20T17:04:12.154156Z","iopub.execute_input":"2023-09-20T17:04:12.154590Z","iopub.status.idle":"2023-09-20T17:04:12.281193Z","shell.execute_reply.started":"2023-09-20T17:04:12.154551Z","shell.execute_reply":"2023-09-20T17:04:12.279751Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## Regroup","metadata":{}},{"cell_type":"code","source":"train_series_file = \"/kaggle/input/child-mind-institute-detect-sleep-states/train_series.parquet\"\n\n# This schema can be optimized.\nparquet_schema = pa.schema(\n    [\n        (\"series_id\", pa.string()),\n        (\"step\", pa.int32()),\n        (\"timestamp\", pa.string()),\n        (\"anglez\", pa.float32()),\n        (\"enmo\", pa.float32()),\n    ]\n)","metadata":{"execution":{"iopub.status.busy":"2023-09-20T17:04:12.283311Z","iopub.execute_input":"2023-09-20T17:04:12.285206Z","iopub.status.idle":"2023-09-20T17:04:12.290824Z","shell.execute_reply.started":"2023-09-20T17:04:12.285170Z","shell.execute_reply":"2023-09-20T17:04:12.289839Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def write_data(input_path, output_path, parquet_schema):\n    parquet_file = pq.ParquetFile(input_path)\n    data = parquet_file.read().to_pandas()\n    grouped_data = data.groupby(\"series_id\")\n\n    writer = pq.ParquetWriter(output_path, parquet_schema)\n\n    for series_id, group in tqdm(grouped_data):\n        table = pa.Table.from_pandas(group, schema=parquet_schema)\n        writer.write_table(table)\n    writer.close()","metadata":{"execution":{"iopub.status.busy":"2023-09-20T17:04:12.292209Z","iopub.execute_input":"2023-09-20T17:04:12.293212Z","iopub.status.idle":"2023-09-20T17:04:12.312002Z","shell.execute_reply.started":"2023-09-20T17:04:12.293173Z","shell.execute_reply":"2023-09-20T17:04:12.310232Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"Alternative if you want to regroup to every 17280 bins (24hr)\n\n```python\ndef write_data(input_path, output_path, parquet_schema, slice_size=17280):\n    parquet_file = pq.ParquetFile(input_path)\n    data = parquet_file.read().to_pandas()\n\n    writer = pq.ParquetWriter(output_path, parquet_schema)\n\n    num_slices = (len(data) + slice_size - 1) // slice_size\n    for slice_idx in tqdm(range(num_slices), desc=\"Writing Slices\"):\n        start_idx = slice_idx * slice_size\n        end_idx = min((slice_idx + 1) * slice_size, len(data))\n        slice_data = data.iloc[start_idx:end_idx]\n\n        table = pa.Table.from_pandas(slice_data, schema=parquet_schema)\n        writer.write_table(table)\n\n    writer.close()\n```","metadata":{}},{"cell_type":"code","source":"%%time\nwrite_data(train_series_file, \"train_series_regrouped.parquet\", parquet_schema)","metadata":{"execution":{"iopub.status.busy":"2023-09-20T17:04:12.314664Z","iopub.execute_input":"2023-09-20T17:04:12.315797Z","iopub.status.idle":"2023-09-20T17:06:51.272794Z","shell.execute_reply.started":"2023-09-20T17:04:12.315756Z","shell.execute_reply":"2023-09-20T17:06:51.271465Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## Loading data from the regrouped data","metadata":{}},{"cell_type":"code","source":"%%time\n# Define the path to the regrouped Parquet file\nregrouped_path = \"train_series_regrouped.parquet\"\nregrouped_file = pq.ParquetFile(regrouped_path)\nnum_row_groups = regrouped_file.num_row_groups\nprint(f\"Number of groups: {num_row_groups}\")","metadata":{"execution":{"iopub.status.busy":"2023-09-20T17:06:51.274496Z","iopub.execute_input":"2023-09-20T17:06:51.274841Z","iopub.status.idle":"2023-09-20T17:06:51.285126Z","shell.execute_reply.started":"2023-09-20T17:06:51.274812Z","shell.execute_reply":"2023-09-20T17:06:51.283663Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def get_series(series_id):\n    table = regrouped_file.read_row_group(series_id)\n    return table.to_pandas()","metadata":{"execution":{"iopub.status.busy":"2023-09-20T17:06:51.286630Z","iopub.execute_input":"2023-09-20T17:06:51.287152Z","iopub.status.idle":"2023-09-20T17:06:51.374621Z","shell.execute_reply.started":"2023-09-20T17:06:51.287110Z","shell.execute_reply":"2023-09-20T17:06:51.373352Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"%%time\ndf = get_series(22)","metadata":{"execution":{"iopub.status.busy":"2023-09-20T17:06:51.376173Z","iopub.execute_input":"2023-09-20T17:06:51.377079Z","iopub.status.idle":"2023-09-20T17:06:51.729492Z","shell.execute_reply.started":"2023-09-20T17:06:51.377048Z","shell.execute_reply":"2023-09-20T17:06:51.728322Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"%%time\ndf","metadata":{"execution":{"iopub.status.busy":"2023-09-20T17:06:51.730917Z","iopub.execute_input":"2023-09-20T17:06:51.731254Z","iopub.status.idle":"2023-09-20T17:06:51.753983Z","shell.execute_reply.started":"2023-09-20T17:06:51.731225Z","shell.execute_reply":"2023-09-20T17:06:51.752691Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"Now, you can create faster and flexible data pipelines.\n\nDon't forget to vote, if you find this notebook useful.","metadata":{}}]}