<jupyter_start><jupyter_text>TensorFlow: Convert TFRecords to Parquet files TFRecords[TFRecords](https://www.tensorflow.org/tutorials/load_data/tfrecord) are a popular file format to store data for deep learning training with TensorFlow. It is a "simple format for storing a sequence of binary records". In many cases the dataset is too large for the host memory and the dataset is converted into (multiple) tfrecords file to disk. TensorFlow's ecosystem enables to stream the tfrecords from disk to train the model without requiring to load the full dataset.That sounds great, but there are some disadvantages when working with tabular dataset. TFRecords stores the dataset as key, values. In other domains, such as computer vision, this representation is efficient as the key is `image` and the values are a the pixels. For an RGB image with 200x200 resolution, there are 120000 (200x200x3) values. In a tabular dataset, a feature is often a single number and therefore, there is a significant overhead for using a key in each example. **In some of our experiments, we experienced that tfrecords can be ~4-5x larger than `parquet` files for the same dataset.**[Parquet](https://en.wikipedia.org/wiki/Apache_Parquet) is another file format to store data. It is a free and open-source data storage format in the Hadoop ecosystem. Many popular systems, such as Spark or Pandas, support to read and write parquet files. We developed [NVTabular Data Loaders](https://nvidia-merlin.github.io/NVTabular/main/training/index.html) as a customized data loader, fully operating on the GPU. It reads the data from disk into the GPU memory and prepares the next batch on the GPU. Therefore, we do not have any CPU-GPU communication. Our data loader leverages parquet files to reduce the disk pressure. **In our experiments, we experienced that the native data loader is the bottleneck in training tabular deep learning models and by changing the native data loader to NVTabular Data Loader, we saw a 8-9x speed-up.** Convert TFRecords to Parquet filesThat is a lot of background information. In many cases, we saw that users have their dataset stored as tfrecords files. In this notebook, we provide a tfrecords to parquet examples. Users can transform their dataset to parquet and be able to experiment with NVTabular data loader. We leverage the library pandas-tfrecords. We install pandas-tfrecords without dependencies, as it would install a specific TensorFlow version.<jupyter_code>!pip install --no-deps pandas-tfrecords==0.1.5
!pip install s3fs<jupyter_output>Looking in indexes: https://pypi.org/simple, https://pypi.ngc.nvidia.com
Requirement already satisfied: pandas-tfrecords==0.1.5 in /usr/local/lib/python3.8/dist-packages (0.1.5)
[33mWARNING: You are using pip version 21.0.1; however, version 21.2.4 is available.
You should consider upgrading via the '/usr/bin/python -m pip install --upgrade pip' command.[0m
Looking in indexes: https://pypi.org/simple, https://pypi.ngc.nvidia.com
Requirement already satisfied: s3fs in /usr/local/lib/python3.8/dist-packages (2021.8.1)
Requirement already satisfied: fsspec==2021.08.1 in /root/.local/lib/python3.8/site-packages/fsspec-2021.8.1-py3.8.egg (from s3fs) (2021.8.1)
Requirement already satisfied: aiobotocore~=1.4.0 in /usr/local/lib/python3.8/dist-packages (from s3fs) (1.4.1)
Requirement already satisfied: wrapt>=1.10.10 in /usr/local/lib/python3.8/dist-packages (from aiobotocore~=1.4.0->s3fs) (1.12.1)
Requirement already satisfied: aioitertools>=0.5.1 in /usr/local/lib/python3.8/dist-packages [...]<jupyter_text>Create a Synthetic Dataset First, we will create a synthetic dataset. Afterwards, we will convert the synthetic data to a tfrecord file. The synthetic dataset contains `continuous features`, `categorical features`, `continuous features in a list with variable length`, `categorical features in a list with variable length` and the `label`.The features of a list have variable length, which are often used in session-based recommender systems. For example, the last page views in a session and sessions have different lengths.<jupyter_code>import numpy as np
import pandas as pd

import cudf
def create_synthetic_df(
    N_CONT_FEATURES, N_CAT_FEATURES, N_CONT_LIST_FEATURES, N_CAT_LIST_FEATURES, N_ROWS
):
    dict_features = {}
    for icont in range(N_CONT_FEATURES):
        dict_features["cont" + str(icont)] = np.random.uniform(-1, 1, size=N_ROWS)
    for icat in range(N_CAT_FEATURES):
        dict_features["cat" + str(icat)] = np.random.choice(list(range(10)), size=N_ROWS)
    for icontlist in range(N_CONT_LIST_FEATURES):
        feature_list = []
        for irow in range(N_ROWS):
            n_elements = np.random.choice(list(range(20)))
            feature_list.append(np.random.uniform(-1, 1, size=n_elements).tolist())
        dict_features["cont_list" + str(icontlist)] = feature_list
    for icatlist in range(N_CAT_LIST_FEATURES):
        feature_list = []
        for irow in range(N_ROWS):
            n_elements = np.random.choice(list(range(20)))
            feature_list.append(np.random.choice(list(range(10)), size=n_elements).tolist())
        dict_features["cat_list" + str(icatlist)] = feature_list
    dict_features["label"] = np.random.choice(list(range(2)), size=N_ROWS)
    df = pd.DataFrame(dict_features)
    return df<jupyter_output><empty_output><jupyter_text>We can configure the size of the dataset and numbers of features of the different type. As this is just a example, we use only 20,000 rows.<jupyter_code>N_ROWS = 20000
N_CONT_FEATURES = 5
N_CAT_FEATURES = 7
N_CONT_LIST_FEATURES = 2
N_CAT_LIST_FEATURES = 3
df = create_synthetic_df(
    N_CONT_FEATURES, N_CAT_FEATURES, N_CONT_LIST_FEATURES, N_CAT_LIST_FEATURES, N_ROWS
)<jupyter_output><empty_output><jupyter_text>We can take a look on the dataset.<jupyter_code>df.head()
CONTINUOUS_COLUMNS = ["cont" + str(i) for i in range(N_CONT_FEATURES)]
CATEGORICAL_COLUMNS = ["cat" + str(i) for i in range(N_CAT_FEATURES)]
CONTINUOUS_LIST_COLUMNS = ["cont_list" + str(i) for i in range(N_CONT_LIST_FEATURES)]
CATEGORICAL_LIST_COLUMNS = ["cat_list" + str(i) for i in range(N_CAT_LIST_FEATURES)]
LABEL_COLUMNS = ["label"]<jupyter_output><empty_output><jupyter_text>Convert the Synthetic Dataset into TFRecords After we created the synthetic dataset, we store it to tfrecords.<jupyter_code>import tensorflow as tf
import os
import multiprocessing as mp
from itertools import repeat


def transform_tfrecords(
    df,
    PATH,
    CONTINUOUS_COLUMNS,
    CATEGORICAL_COLUMNS,
    CONTINUOUS_LIST_COLUMNS,
    CATEGORICAL_LIST_COLUMNS,
    LABEL_COLUMNS,
):
    write_dir = os.path.dirname(PATH)
    if not os.path.exists(write_dir):
        os.makedirs(write_dir)
    file_idx, example_idx = 0, 0
    writer = get_writer(write_dir, file_idx)
    column_names = [
        CONTINUOUS_COLUMNS,
        CATEGORICAL_COLUMNS + LABEL_COLUMNS,
        CONTINUOUS_LIST_COLUMNS,
        CATEGORICAL_LIST_COLUMNS,
    ]
    with mp.Pool(8, pool_initializer, column_names) as pool:
        data = []
        for col_names in column_names:
            if len(col_names) == 0:
                data.append(repeat(None))
            else:
                data.append(df[col_names].values)
        data = zip(*data)
        record_map = pool.imap(build_and_serialize_example, data, chunksize=200)
        for record in record_map:
            writer.write(record)
            example_idx += 1
    writer.close()


def pool_initializer(num_cols, cat_cols, num_list_cols, cat_list_cols):
    global numeric_columns
    global categorical_columns
    global numeric_list_columns
    global categorical_list_columns
    numeric_columns = num_cols
    categorical_columns = cat_cols
    numeric_list_columns = num_list_cols
    categorical_list_columns = cat_list_cols


def build_and_serialize_example(data):
    numeric_values, categorical_values, numeric_list_values, categorical_list_values = data
    feature = {}
    if numeric_values is not None:
        feature.update(
            {
                col: tf.train.Feature(float_list=tf.train.FloatList(value=[val]))
                for col, val in zip(numeric_columns, numeric_values)
            }
        )
    if categorical_values is not None:
        feature.update(
            {
                col: tf.train.Feature(int64_list=tf.train.Int64List(value=[val]))
                for col, val in zip(categorical_columns, categorical_values)
            }
        )
    if numeric_list_values is not None:
        feature.update(
            {
                col: tf.train.Feature(float_list=tf.train.FloatList(value=val))
                for col, val in zip(numeric_list_columns, numeric_list_values)
            }
        )
    if categorical_list_values is not None:
        feature.update(
            {
                col: tf.train.Feature(int64_list=tf.train.Int64List(value=val))
                for col, val in zip(categorical_list_columns, categorical_list_values)
            }
        )
    return tf.train.Example(features=tf.train.Features(feature=feature)).SerializeToString()


def get_writer(write_dir, file_idx):
    filename = str(file_idx).zfill(5) + ".tfrecords"
    return tf.io.TFRecordWriter(os.path.join(write_dir, filename))<jupyter_output><empty_output><jupyter_text>We define the output path.<jupyter_code>PATH = "/raid/tfrecord-test/"
!rm -rf $PATH
!mkdir $PATH
transform_tfrecords(
    df,
    PATH,
    CONTINUOUS_COLUMNS,
    CATEGORICAL_COLUMNS,
    CONTINUOUS_LIST_COLUMNS,
    CATEGORICAL_LIST_COLUMNS,
    LABEL_COLUMNS,
)<jupyter_output><empty_output><jupyter_text>We can check the file.<jupyter_code>!ls $PATH<jupyter_output>00000.tfrecords<jupyter_text>Convert TFRecords to parquet files Now, we have a dataset in the tfrecords format. Let's use the `convert_tfrecords_to_parquet` function to convert a tfrecord file into parquet.<jupyter_code>import glob

from nvtabular.framework_utils.tensorflow.tfrecords_to_parquet import convert_tfrecords_to_parquet<jupyter_output><empty_output><jupyter_text>Let's select all TFRecords in the folder.<jupyter_code>filenames = glob.glob(PATH + "/*.tfrecords")<jupyter_output><empty_output><jupyter_text>Let's call the `convert_tfrecords_to_parquet`.Some details about the parameters:* `compression_type` is the compression type of the tfrecords. Options: `""` (no compression), `"ZLIB"`, or `"GZIP"`* `chunks` defines how many data points per `parquet` file should be saved. It splits a tfrecords into multiple parquet files.* `convert_lists` defines, if feature lists should be converted into muliple feature columns. Even single dataframe series are 1 dimensional arrays when converted back from tfrecords to parquet.<jupyter_code>filenames
convert_tfrecords_to_parquet(
    filenames=filenames, output_dir=PATH, compression_type="", chunks=1000, convert_lists=True
)<jupyter_output>2021-09-22 21:56:53.202269: I tensorflow/core/platform/cpu_feature_guard.cc:142] This TensorFlow binary is optimized with oneAPI Deep Neural Network Library (oneDNN) to use the following CPU instructions in performance-critical operations:  AVX2 FMA
To enable them in other operations, rebuild TensorFlow with the appropriate compiler flags.
2021-09-22 21:56:54.586055: I tensorflow/core/common_runtime/gpu/gpu_device.cc:1510] Created device /job:localhost/replica:0/task:0/device:GPU:0 with 30681 MB memory:  -> device: 0, name: Tesla V100-SXM2-32GB, pci bus id: 0000:0b:00.0, compute capability: 7.0
2021-09-22 21:56:55.158643: I tensorflow/compiler/mlir/mlir_graph_optimization_pass.cc:185] None of the MLIR Optimization Passes are enabled (registered 2)
20000it [00:12, 1665.20it/s]<jupyter_text>Let's take a look We can see that `convert_tfrecords_to_parquet` created multiple files per `tfrecord` depending on the chunk size.<jupyter_code>filenames = glob.glob(PATH + "/*.parquet")
filenames<jupyter_output><empty_output><jupyter_text>If we load the first file, we can see, that it has the same structure as our original synthetic dataset.<jupyter_code>df = cudf.read_parquet(filenames[0])
df.head()
df.shape<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.08622398414271557, 'fraction_numerical': 0.02271225635943178, 'mean_word_length': 4.299343544857768, 'pattern_counts': {'":': 0, '<': 46, '<?xml version=': 0, '>': 50, 'https://': 7, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '12', 'dir': 'jupyter-structured-clean-dedup', 'id': '112649', 'license': 'Apache-2.0', 'n_tokens_mistral': 3999, 'n_tokens_neox': 3773, 'n_words': 1128, 'path': 'examples/tensorflow/TFRecords-To-Parquet.ipynb', 'repo_name': 'thibaultcharrin/NVTabular', 'size': '12108'}