{"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":"# How to load data, perform ragged batching and save\n## Training data generated by the methods in this notebook have the following benefits\n* No information lost, even the longest sequences can easily be saved\n* Used disk space is minimized, because no padding of the sequences is needed\n* Batching improves training speed and performance\n* Defult data format is consistent with the data delivery of submission evaluation (up to batching)\n* Your custom preprocessing functions are easy to add!","metadata":{}},{"cell_type":"markdown","source":"# Imports and constants","metadata":{}},{"cell_type":"code","source":"import tensorflow as tf\nimport pandas as pd\nimport numpy as np\nimport json\nfrom tqdm import tqdm","metadata":{"execution":{"iopub.status.busy":"2023-03-05T09:45:04.362884Z","iopub.execute_input":"2023-03-05T09:45:04.363229Z","iopub.status.idle":"2023-03-05T09:45:07.730392Z","shell.execute_reply.started":"2023-03-05T09:45:04.363194Z","shell.execute_reply":"2023-03-05T09:45:07.729199Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"with open('/kaggle/input/asl-signs/sign_to_prediction_index_map.json') as f:\n    sign_ids = json.load(f)\nprint(sign_ids)","metadata":{"scrolled":true,"execution":{"iopub.status.busy":"2023-03-05T10:22:35.444233Z","iopub.execute_input":"2023-03-05T10:22:35.446249Z","iopub.status.idle":"2023-03-05T10:22:35.461553Z","shell.execute_reply.started":"2023-03-05T10:22:35.446168Z","shell.execute_reply":"2023-03-05T10:22:35.460031Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"DATA_COLUMNS    = ['x', 'y', 'z']\nROWS_PER_FRAME  = 543\nNUM_SHARDS      = 2\nSAVE_PATH       = '/tmp/GoogleISLDataset'\nBATCH_SIZE      = 256","metadata":{"execution":{"iopub.status.busy":"2023-03-05T09:45:07.753186Z","iopub.execute_input":"2023-03-05T09:45:07.753540Z","iopub.status.idle":"2023-03-05T09:45:07.758552Z","shell.execute_reply.started":"2023-03-05T09:45:07.753506Z","shell.execute_reply":"2023-03-05T09:45:07.757668Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"# Data loading functions\n* Notice that this returns data in the competition format!","metadata":{}},{"cell_type":"code","source":"def load_relevant_data_subset(pq_path):\n    data = pd.read_parquet('/kaggle/input/asl-signs/'+pq_path, columns=DATA_COLUMNS)\n    n_frames = int(len(data) / ROWS_PER_FRAME)\n    data = data.values.astype(np.float32)\n    return data.reshape(n_frames, ROWS_PER_FRAME, len(DATA_COLUMNS))","metadata":{"execution":{"iopub.status.busy":"2023-03-05T09:45:07.761640Z","iopub.execute_input":"2023-03-05T09:45:07.762214Z","iopub.status.idle":"2023-03-05T09:45:07.769170Z","shell.execute_reply.started":"2023-03-05T09:45:07.762178Z","shell.execute_reply":"2023-03-05T09:45:07.768108Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"TensorFlow Dataset.map() only accepts graph traceable functions, so we wrap the data loading function with tf.py_function","metadata":{}},{"cell_type":"code","source":"def tf_get_features(ftensor):\n    def feat_wrapper(ftensor):\n        return load_relevant_data_subset(ftensor.numpy().decode('utf-8'))\n    return tf.py_function(\n        feat_wrapper,\n        [ftensor],\n        Tout=tf.float32\n    )","metadata":{"execution":{"iopub.status.busy":"2023-03-05T09:45:07.770582Z","iopub.execute_input":"2023-03-05T09:45:07.770968Z","iopub.status.idle":"2023-03-05T09:45:07.786918Z","shell.execute_reply.started":"2023-03-05T09:45:07.770931Z","shell.execute_reply":"2023-03-05T09:45:07.785622Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"tf.py_function let's us execute code outside the tf graph, but that means TensorFlow cannot track the shapes of the returned tensors!\n\nHowever, we still know the returned shape from looking at the loading function, so we can use tf.ensure_shape to tell the rest of our pipeline what element shapes to expect. This is necessary for the dense_to_ragged_batch function that will finally perform the batching","metadata":{}},{"cell_type":"code","source":"def set_shape(x):\n    \n    # None dimensions can be of any length\n    return tf.ensure_shape(x, (None, ROWS_PER_FRAME, len(DATA_COLUMNS)))","metadata":{"execution":{"iopub.status.busy":"2023-03-05T09:45:07.788357Z","iopub.execute_input":"2023-03-05T09:45:07.788829Z","iopub.status.idle":"2023-03-05T09:45:07.797748Z","shell.execute_reply.started":"2023-03-05T09:45:07.788781Z","shell.execute_reply":"2023-03-05T09:45:07.796693Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"# Generating the dataset\n\nHere it would be possible to also apply your own preprocessing functions! For most flexibility, apply preprocessing before batching. If your preprocessing function can process RaggedTensors, you can likely save the data faster by  preprocessing after batching","metadata":{}},{"cell_type":"code","source":"train_df = pd.read_csv('/kaggle/input/asl-signs/train.csv')","metadata":{"execution":{"iopub.status.busy":"2023-03-05T09:45:07.799070Z","iopub.execute_input":"2023-03-05T09:45:07.800037Z","iopub.status.idle":"2023-03-05T09:45:07.930222Z","shell.execute_reply.started":"2023-03-05T09:45:07.799999Z","shell.execute_reply":"2023-03-05T09:45:07.928893Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"X_ds = tf.data.Dataset.from_tensor_slices(\n    train_df.path.values                                              # start with a dataset of the parquet paths\n).map(\n    tf_get_features                                                   # load individual sequences\n).map(\n    set_shape                                                         # set and enforce element shape\n).apply(\n    tf.data.experimental.dense_to_ragged_batch(batch_size=BATCH_SIZE) # apply batching function\n)\n\n# load and batch the labels\ny_ds = tf.data.Dataset.from_tensor_slices(\n    train_df.sign.map(sign_ids).values.reshape(-1,1)\n).batch(BATCH_SIZE)\n\n# zip the features and labels\ntrain_ds = tf.data.Dataset.zip((X_ds, y_ds))","metadata":{"execution":{"iopub.status.busy":"2023-03-05T09:45:07.932061Z","iopub.execute_input":"2023-03-05T09:45:07.932870Z","iopub.status.idle":"2023-03-05T09:45:08.123832Z","shell.execute_reply.started":"2023-03-05T09:45:07.932796Z","shell.execute_reply":"2023-03-05T09:45:08.122621Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"# Saving\nSaved Datasets use sharding, which means data is saved to several files to enable parallel reading. By default, TensorFlow decides the number of shards by the number of CPU cores available, but that may not be desirable so we use a custom sharding function to divide the batches into NUM_SHARDS files.","metadata":{}},{"cell_type":"code","source":"# Sharding could be improved, as the distribution of elements in different shards should optimally be equal.\n# Currently, it will be a sample from a uniform distribution because this is simple to implement\ndef shard_func(*_):\n    return tf.random.uniform(shape=[], maxval=NUM_SHARDS, dtype=tf.int64)","metadata":{"execution":{"iopub.status.busy":"2023-03-05T09:45:08.125846Z","iopub.execute_input":"2023-03-05T09:45:08.126287Z","iopub.status.idle":"2023-03-05T09:45:08.132137Z","shell.execute_reply.started":"2023-03-05T09:45:08.126239Z","shell.execute_reply":"2023-03-05T09:45:08.130910Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"train_ds.prefetch(tf.data.AUTOTUNE).save(SAVE_PATH, shard_func=shard_func)","metadata":{"execution":{"iopub.status.busy":"2023-03-05T09:45:08.133634Z","iopub.execute_input":"2023-03-05T09:45:08.134032Z","iopub.status.idle":"2023-03-05T10:12:06.966787Z","shell.execute_reply.started":"2023-03-05T09:45:08.133995Z","shell.execute_reply":"2023-03-05T10:12:06.964803Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"# Check speed for reading from disk","metadata":{}},{"cell_type":"code","source":"def check_throughput(ds_path):\n    for x in tqdm(tf.data.Dataset.load(ds_path)):\n        pass","metadata":{"execution":{"iopub.status.busy":"2023-03-05T10:12:06.970179Z","iopub.execute_input":"2023-03-05T10:12:06.970859Z","iopub.status.idle":"2023-03-05T10:12:06.977638Z","shell.execute_reply.started":"2023-03-05T10:12:06.970783Z","shell.execute_reply":"2023-03-05T10:12:06.976417Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"check_throughput(SAVE_PATH)","metadata":{"execution":{"iopub.status.busy":"2023-03-05T10:12:06.979394Z","iopub.execute_input":"2023-03-05T10:12:06.979745Z","iopub.status.idle":"2023-03-05T10:14:43.003301Z","shell.execute_reply.started":"2023-03-05T10:12:06.979710Z","shell.execute_reply":"2023-03-05T10:14:42.999723Z"},"trusted":true},"execution_count":null,"outputs":[]}]}