Source code for merlin.io.dataset

#
# Copyright (c) 2022, NVIDIA CORPORATION.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

import collections
import logging
import math
import random
import warnings
from pathlib import Path

import dask
import distributed
import numpy as np
from dask.base import tokenize
from dask.dataframe.core import new_dd_object
from dask.highlevelgraph import HighLevelGraph
from dask.utils import natural_sort_key, parse_bytes
from fsspec.core import get_fs_token_paths
from fsspec.utils import stringify_path
from npy_append_array import NpyAppendArray

from merlin.core.compat import HAS_GPU, cudf, dask_cudf, device_mem_size
from merlin.core.dispatch import (
    convert_data,
    dataframe_columnwise_explode,
    hex_to_int,
    is_dataframe_object,
    is_list_dtype,
    list_val_dtype,
)
from merlin.core.utils import global_dask_client, set_client_deprecated
from merlin.dtypes.shape import DefaultShapes
from merlin.io.csv import CSVDatasetEngine
from merlin.io.dask import _ddf_to_dataset, _simple_shuffle
from merlin.io.dataframe_engine import DataFrameDatasetEngine
from merlin.io.dataframe_iter import DataFrameIter
from merlin.io.parquet import ParquetDatasetEngine
from merlin.io.shuffle import _check_shuffle_arg
from merlin.schema import ColumnSchema, Schema
from merlin.schema.io.tensorflow_metadata import TensorflowMetadata

MERLIN_METADATA_DIR_NAME = ".merlin"
LOG = logging.getLogger("merlin")


[docs] class Dataset: """Universal external-data wrapper for NVTabular The NVTabular `Workflow` and `DataLoader`-related APIs require all external data to be converted to the universal `Dataset` type. The main purpose of this class is to abstract away the raw format of the data, and to allow other NVTabular classes to reliably materialize a `dask_cudf.DataFrame` collection (and/or collection-based iterator) on demand. A new `Dataset` object can be initialized from a variety of different raw-data formats. To initialize an object from a directory path or file list, the `engine` argument should be used to specify either "parquet" or "csv" format. If the first argument contains a list of files with a suffix of either "parquet" or "csv", the engine can be inferred:: # Initialize Dataset with a parquet-dataset directory. # must specify engine="parquet" dataset = Dataset("/path/to/data_pq", engine="parquet") # Initialize Dataset with list of csv files. # engine="csv" argument is optional dataset = Dataset(["file_0.csv", "file_1.csv"]) Since NVTabular leverages `fsspec` as a file-system interface, the underlying data can be stored either locally, or in a remote/cloud data store. To read from remote storage, like gds or s3, the appropriate protocol should be prepended to the `Dataset` path argument(s), and any special backend parameters should be passed in a `storage_options` dictionary:: # Initialize Dataset with s3 parquet data dataset = Dataset( "s3://bucket/path", engine="parquet", storage_options={'anon': True, 'use_ssl': False}, ) By default, both parquet and csv-based data will be converted to a Dask-DataFrame collection with a maximum partition size of roughly 12.5 percent of the total memory on a single device. The partition size can be changed to a different fraction of total memory on a single device with the `part_mem_fraction` argument. Alternatively, a specific byte size can be specified with the `part_size` argument:: # Dataset partitions will be ~10% single-GPU memory (or smaller) dataset = Dataset("bigfile.parquet", part_mem_fraction=0.1) # Dataset partitions will be ~1GB (or smaller) dataset = Dataset("bigfile.parquet", part_size="1GB") Note that, if both the fractional and literal options are used at the same time, `part_size` will take precedence. Also, for parquet-formatted data, the partitioning is done at the row- group level, and the byte-size of the first row-group (after CuDF conversion) is used to map all other partitions. Therefore, if the distribution of row-group sizes is not uniform, the partition sizes will not be balanced. In addition to handling data stored on disk, a `Dataset` object can also be initialized from an existing CuDF/Pandas DataFrame, or from a Dask-DataFrame collection (e.g. `dask_cudf.DataFrame`). For these in-memory formats, the size/number of partitions will not be modified. That is, a CuDF/Pandas DataFrame (or PyArrow Table) will produce a single-partition collection, while the number/size of a Dask-DataFrame collection will be preserved:: # Initialize from CuDF DataFrame (creates 1 partition) gdf = cudf.DataFrame(...) dataset = Dataset(gdf) # Initialize from Dask-CuDF DataFrame (preserves partitions) ddf = dask_cudf.read_parquet(...) dataset = Dataset(ddf) Since the `Dataset` API can both ingest and output a Dask collection, it is straightforward to transform data either before or after an NVTabular workflow is executed. This means that some complex pre-processing operations, that are not yet supported in NVTabular, can still be accomplished with the Dask-CuDF API:: # Sort input data before final Dataset initialization # Warning: Global sorting requires significant device memory! ddf = Dataset("/path/to/data_pq", engine="parquet").to_ddf() ddf = ddf.sort_values("user_rank", ignore_index=True) dataset = Dataset(ddf) `Dataset Optimization Tips (DOTs)` The NVTabular dataset should be created from Parquet files in order to get the best possible performance, preferably with a row group size of around 128MB. While NVTabular also supports reading from CSV files, reading CSV can be over twice as slow as reading from Parquet. Take a look at this notebook_ for an example of transforming the original Criteo CSV dataset into a new Parquet dataset optimized for use with NVTabular. .. _notebook: https://github.com/NVIDIA/NVTabular/blob/main/examples/optimize_criteo.ipynb Parameters ----------- path_or_source : str, list of str, or <dask.dataframe|cudf|pd>.DataFrame Dataset path (or list of paths), or a DataFrame. If string, should specify a specific file or directory path. If this is a directory path, the directory structure must be flat (nested directories are not yet supported). engine : str or DatasetEngine DatasetEngine object or string identifier of engine. Current string options include: ("parquet", "csv", "avro"). This argument is ignored if path_or_source is a DataFrame type. npartitions : int Desired number of Dask-collection partitions to produce in the ``to_ddf`` method when ``path_or_source`` corresponds to a DataFrame type. This argument is ignored for file-based ``path_or_source`` input. part_size : str or int Desired size (in bytes) of each Dask partition. If None, part_mem_fraction will be used to calculate the partition size. Note that the underlying engine may allow other custom kwargs to override this argument. This argument is ignored if path_or_source is a DataFrame type. part_mem_fraction : float (default 0.125) Fractional size of desired dask partitions (relative to GPU memory capacity). Ignored if part_size is passed directly. Note that the underlying engine may allow other custom kwargs to override this argument. This argument is ignored if path_or_source is a DataFrame type. If ``cpu=True``, this value will be relative to the total host memory detected by the client process. storage_options: None or dict Further parameters to pass to the bytes backend. This argument is ignored if path_or_source is a DataFrame type. cpu : bool WARNING: Experimental Feature! Whether NVTabular should keep all data in cpu memory when the Dataset is converted to an internal Dask collection. The default value is False, unless ``cudf`` and ``dask_cudf`` are not installed (in which case the default is True). In the future, if True, NVTabular will NOT use any available GPU devices for down-stream processing. NOTE: Down-stream ops and output do not yet support a Dataset generated with ``cpu=True``. base_dataset : Dataset Optional reference to the original "base" Dataset object used to construct the current Dataset instance. This object is used to preserve file-partition mapping information. schema : Schema Optional argument, to support custom user defined Schemas. This overrides the derived schema behavior. **kwargs : Key-word arguments to pass through to Dask.dataframe IO function. For the Parquet engine(s), notable arguments include `filters` and `aggregate_files` (the latter is experimental). """
[docs] def __init__( self, path_or_source, engine=None, npartitions=None, part_size=None, part_mem_fraction=None, storage_options=None, dtypes=None, client="auto", cpu=None, base_dataset=None, schema=None, **kwargs, ): if schema is not None and not isinstance(schema, Schema): raise TypeError(f"unsupported schema type for merlin.io.Dataset: {type(schema)}") # Deprecate `client` if client != "auto": set_client_deprecated(client, "Dataset") self.dtypes = dtypes self.schema = schema # Cache for "real" (sampled) metadata self._real_meta = {} # Check if we are keeping data in host or gpu device memory self.cpu = cpu if self.cpu is False: if not HAS_GPU: raise RuntimeError( "Cannot initialize Dataset on GPU. " "No devices detected (with pynvml). " "Check that pynvml can be initialized. " ) if cudf is None: raise RuntimeError( "Cannot initialize Dataset on GPU. " "cudf package not found. " "Check that cudf is installed in this environment and can be imported. " ) if self.cpu is None: self.cpu = cudf is None or not HAS_GPU # Keep track of base dataset (optional) self.base_dataset = base_dataset or self # For now, lets warn the user that "cpu mode" is experimental if self.cpu: warnings.warn( "Initializing an NVTabular Dataset in CPU mode." "This is an experimental feature with extremely limited support!" ) npartitions = npartitions or 1 if isinstance(path_or_source, dask.dataframe.DataFrame) or is_dataframe_object( path_or_source ): # User is passing in a <dask.dataframe|cudf|pd>.DataFrame # Use DataFrameDatasetEngine _path_or_source = convert_data( path_or_source, cpu=self.cpu, to_collection=True, npartitions=npartitions ) # Check if this is a collection that has now moved between host <-> device moved_collection = isinstance(path_or_source, dask.dataframe.DataFrame) and ( not isinstance(_path_or_source._meta, type(path_or_source._meta)) ) if part_size: warnings.warn("part_size is ignored for DataFrame input.") if part_mem_fraction: warnings.warn("part_mem_fraction is ignored for DataFrame input.") self.engine = DataFrameDatasetEngine( _path_or_source, cpu=self.cpu, moved_collection=moved_collection ) else: if part_size: # If a specific partition size is given, use it directly part_size = parse_bytes(part_size) else: # If a fractional partition size is given, calculate part_size part_mem_fraction = part_mem_fraction or 0.125 assert 0.0 < part_mem_fraction < 1.0 if part_mem_fraction > 0.25: warnings.warn( "Using very large partitions sizes for Dask. " "Memory-related errors are likely." ) part_size = int(device_mem_size(kind="total", cpu=self.cpu) * part_mem_fraction) # Engine-agnostic path handling paths = stringify_path(path_or_source) if isinstance(paths, str): paths = [paths] paths = sorted(paths, key=natural_sort_key) storage_options = storage_options or {} # If engine is not provided, try to infer from end of paths[0] if engine is None: engine = paths[0].split(".")[-1] if isinstance(engine, str): if engine == "parquet": self.engine = ParquetDatasetEngine( paths, part_size, storage_options=storage_options, cpu=self.cpu, **kwargs ) elif engine == "csv": self.engine = CSVDatasetEngine( paths, part_size, storage_options=storage_options, cpu=self.cpu, **kwargs ) elif engine == "avro": try: from merlin.io.avro import AvroDatasetEngine except ImportError as e: raise RuntimeError( "Failed to import AvroDatasetEngine. Make sure uavro is installed." ) from e self.engine = AvroDatasetEngine( paths, part_size, storage_options=storage_options, cpu=self.cpu, **kwargs ) else: raise ValueError("Only parquet, csv, and avro supported (for now).") else: self.engine = engine( paths, part_size, cpu=self.cpu, storage_options=storage_options ) # load in schema or infer if not available # path is always a list at this point if not self.schema: if isinstance(path_or_source, (str, Path)): path_or_source = [Path(path_or_source)] if isinstance(path_or_source, list) and isinstance(path_or_source[0], (str, Path)): # list of paths to files schema_path = Path(path_or_source[0]) if schema_path.is_file(): schema_path = schema_path.parent pbtxt_deprecated_warning = ( "Found schema.pbtxt. Loading schema automatically from " "schema.pbtxt is deprecated and will be removed in the " "future. Re-run workflow to generate .merlin/schema.json." ) if (schema_path / MERLIN_METADATA_DIR_NAME / "schema.json").exists(): schema = TensorflowMetadata.from_json_file( schema_path / MERLIN_METADATA_DIR_NAME ) self.schema = schema.to_merlin_schema() elif (schema_path.parent / MERLIN_METADATA_DIR_NAME / "schema.json").exists(): schema = TensorflowMetadata.from_json_file( schema_path.parent / MERLIN_METADATA_DIR_NAME ) self.schema = schema.to_merlin_schema() elif (schema_path / "schema.pbtxt").exists(): warnings.warn(pbtxt_deprecated_warning, DeprecationWarning) schema = TensorflowMetadata.from_proto_text_file(schema_path) self.schema = schema.to_merlin_schema() elif (schema_path.parent / "schema.pbtxt").exists(): warnings.warn(pbtxt_deprecated_warning, DeprecationWarning) schema = TensorflowMetadata.from_proto_text_file(schema_path.parent) self.schema = schema.to_merlin_schema() else: self.infer_schema() else: # df with no schema self.infer_schema()
[docs] def to_ddf(self, columns=None, shuffle=False, seed=None): """Convert `Dataset` object to `dask_cudf.DataFrame` Parameters ----------- columns : str or list(str); default None Columns to include in output `DataFrame`. If not specified, the output will contain all known columns in the Dataset. shuffle : bool; default False Whether to shuffle the order of partitions in the output `dask_cudf.DataFrame`. Note that this does not shuffle the rows within each partition. This is because the data is not actually loaded into memory for this operation. seed : int; Optional The random seed to use if `shuffle=True`. If nothing is specified, the current system time will be used by the `random` std library. """ # Use DatasetEngine to create ddf ddf = self.engine.to_ddf(columns=columns) # Shuffle the partitions of ddf (optional) if shuffle and ddf.npartitions > 1: # Start with ordered partitions inds = list(range(ddf.npartitions)) # Use random std library to reorder partitions random.seed(seed) random.shuffle(inds) # Construct new high-level graph (HLG) name = ddf._name new_name = "shuffle-partitions-" + tokenize(ddf) dsk = {(new_name, i): (lambda x: x, (name, ind)) for i, ind in enumerate(inds)} new_graph = HighLevelGraph.from_collections(new_name, dsk, dependencies=[ddf]) # Convert the HLG to a Dask collection divisions = [None] * (ddf.npartitions + 1) ddf = new_dd_object(new_graph, new_name, ddf._meta, divisions) # Special dtype conversion (optional) if self.dtypes: _meta = _set_dtypes(ddf._meta, self.dtypes) ddf = ddf.map_partitions(_set_dtypes, self.dtypes, meta=_meta) dask_client = global_dask_client() if dask_client is not None: # pylint: disable=unidiomatic-typecheck if ( dask_cudf and isinstance(ddf, dask_cudf.DataFrame) and type(dask_client.cluster) is distributed.LocalCluster ): raise RuntimeError( "`dask_cudf.DataFrame` is incompatible with `distributed.LocalCluster`. " "Please setup a `dask_cuda.LocalCUDACluster` instead. " "Or to run on CPU instead, " "provide the parameter `cpu=True` when creating the `Dataset`. " ) return ddf
@property def file_partition_map(self): return self.engine._file_partition_map @property def partition_lens(self): return self.engine._partition_lens
[docs] def to_cpu(self): warnings.warn( "Changing an NVTabular Dataset to CPU mode." "This is an experimental feature with extremely limited support!" ) self.cpu = True self.engine.to_cpu()
[docs] def to_gpu(self): self.cpu = False self.engine.to_gpu()
[docs] def shuffle_by_keys(self, keys, hive_data=None, npartitions=None): """Shuffle the in-memory Dataset so that all unique-key combinations are moved to the same partition. Parameters ---------- keys : list(str) Column names to shuffle by. hive_data : bool; default None Whether the dataset is backed by a hive-partitioned dataset (with the keys encoded in the directory structure). By default, the Dataset's `file_partition_map` property will be inspected to infer this setting. When `hive_data` is True, the number of output partitions will correspond to the number of unique key combinations in the dataset. npartitions : int; default None Number of partitions in the output Dataset. For hive-partitioned data, this value should be <= the number of unique key combinations (the default), otherwise it will be ignored. For data that is not hive-partitioned, the ``npartitions`` input should be <= the original partition count, otherwise it will be ignored. """ # Make sure we are dealing with a list keys = [keys] if not isinstance(keys, (list, tuple)) else keys # Start with default ddf ddf = self.to_ddf() if npartitions: npartitions = min(ddf.npartitions, npartitions) if hive_data is not False: # The keys may be encoded in the directory names. # Let's use the file_partition_map to extract this info. try: _mapping = self.file_partition_map except AttributeError as e: _mapping = None if hive_data: raise RuntimeError("Failed to extract hive-partition mapping!") from e # If we have a `_mapping` available, check if the # file names include information about all our keys hive_mapping = collections.defaultdict(list) if _mapping: for k, v in _mapping.items(): for part in k.split(self.engine.fs.sep)[:-1]: try: _key, _val = part.split("=") except ValueError: continue if _key in keys: hive_mapping[_key].append(_val) if set(hive_mapping.keys()) == set(keys): # Generate hive-mapping DataFrame summary hive_mapping = type(ddf._meta)(hive_mapping) cols = list(hive_mapping.columns) for c in keys: typ = ddf._meta[c].dtype if c in cols: if typ == "category": # Cannot cast directly to categorical unless we # first cast to the underlying dtype of the categories hive_mapping[c] = hive_mapping[c].astype(typ.categories.dtype) hive_mapping[c] = hive_mapping[c].astype(typ) # Generate simple-shuffle plan target_mapping = hive_mapping.drop_duplicates().reset_index(drop=True) target_mapping.index.name = "_partition" hive_mapping.index.name = "_sort" target_mapping.reset_index(drop=False, inplace=True) plan = ( hive_mapping.reset_index() .merge(target_mapping, on=cols, how="left") .sort_values("_sort")["_partition"] ) if hasattr(plan, "to_pandas"): plan = plan.to_pandas() # Deal with repartitioning if npartitions and npartitions < len(target_mapping): q = np.linspace(0.0, 1.0, num=npartitions + 1) divs = plan.quantile(q) partitions = divs.searchsorted(plan, side="right") - 1 partitions[(plan >= divs.iloc[-1]).values] = len(divs) - 2 plan = partitions.tolist() elif len(plan) != len(plan.unique()): plan = plan.to_list() else: # Plan is a unique 1:1 ddf partition mapping. # We already have shuffled data. return self # TODO: We should avoid shuffling the original ddf and # instead construct a new (more-efficent) graph to read # multiple files from each partition directory at once. # Generally speaking, we can optimize this code path # much further. return Dataset(_simple_shuffle(ddf, plan)) # Fall back to dask.dataframe algorithm return Dataset(ddf.shuffle(keys, npartitions=npartitions))
[docs] def repartition(self, npartitions=None, partition_size=None): """Repartition the underlying ddf, and return a new Dataset Parameters ---------- npartitions : int; default None Number of partitions in output ``Dataset``. Only used if ``partition_size`` isn’t specified. partition_size : int or str; default None Max number of bytes of memory for each partition. Use numbers or strings like '5MB'. If specified, ``npartitions`` will be ignored. """ return Dataset( self.to_ddf() .clear_divisions() .repartition( npartitions=npartitions, partition_size=partition_size, ), schema=self.schema, cpu=self.cpu, )
[docs] @classmethod def merge(cls, left, right, **kwargs): """Merge two Dataset objects Produces a new Dataset object. If the ``cpu`` Dataset attributes do not match, the right side will be modified. See Dask-Dataframe ``merge`` documentation for more information. Example usage:: ds_1 = Dataset("file.parquet") ds_2 = Dataset(cudf.DataFrame(...)) ds_merged = Dataset.merge(ds_1, ds_2, on="foo", how="inner") Parameters ---------- left : Dataset Left-side Dataset object. right : Dataset Right-side Dataset object. **kwargs : Key-word arguments to be passed through to Dask-Dataframe. """ # Ensure both Dataset objects are either cudf or pandas based if left.cpu and not right.cpu: _right = cls(right.to_ddf()) _right.to_cpu() elif not left.cpu and right.cpu: _right = cls(right.to_ddf()) _right.to_gpu() elif left.cpu == right.cpu: # both left and right are already cudf / pandas df _right = right return cls( left.to_ddf() .clear_divisions() .merge( _right.to_ddf().clear_divisions(), **kwargs, ) )
[docs] def to_iter( self, columns=None, indices=None, shuffle=False, seed=None, use_file_metadata=None, epochs=1 ): """Convert `Dataset` object to a `cudf.DataFrame` iterator. Note that this method will use `to_ddf` to produce a `dask_cudf.DataFrame`, and materialize a single partition for each iteration. Parameters ---------- columns : str or list(str); default None Columns to include in each `DataFrame`. If not specified, the outputs will contain all known columns in the Dataset. indices : list(int); default None A specific list of partition indices to iterate over. If nothing is specified, all partitions will be returned in order (or the shuffled order, if `shuffle=True`). shuffle : bool; default False Whether to shuffle the order of `dask_cudf.DataFrame` partitions used by the iterator. If the `indices` argument is specified, those indices correspond to the partition indices AFTER the shuffle operation. seed : int; Optional The random seed to use if `shuffle=True`. If nothing is specified, the current system time will be used by the `random` std library. use_file_metadata : bool; Optional Whether to allow the returned ``DataFrameIter`` object to use file metadata from the ``base_dataset`` to estimate the row-count. By default, the file-metadata optimization will only be used if the current Dataset is backed by a file-based engine. Otherwise, it is possible that an intermediate transform has modified the row-count. epochs : int Number of dataset passes to include within a single iterator. This option is used for multi-epoch data-loading. Default is 1. """ if isinstance(columns, str): columns = [columns] # Try to extract the row-size metadata # if we are not shuffling partition_lens_meta = None if not shuffle and use_file_metadata is not False: # We are allowed to use file metadata to calculate # partition sizes. If `use_file_metadata` is None, # we only use metadata if `self` is backed by a # file-based engine (like "parquet"). Otherwise, # we cannot be "sure" that the metadata row-count # is correct. try: if use_file_metadata: partition_lens_meta = self.base_dataset.partition_lens else: partition_lens_meta = self.partition_lens except AttributeError: pass return DataFrameIter( self.to_ddf(columns=columns, shuffle=shuffle, seed=seed), indices=indices, partition_lens=partition_lens_meta, epochs=epochs, )
[docs] def to_parquet( self, output_path, shuffle=None, preserve_files=False, output_files=None, out_files_per_proc=None, row_group_size=None, num_threads=0, dtypes=None, cats=None, conts=None, labels=None, suffix=".parquet", partition_on=None, method="subgraph", write_hugectr_keyset=False, ): """Writes out to a parquet dataset Parameters ---------- output_path : string Path to write processed/shuffled output data shuffle : merlin.io.Shuffle enum How to shuffle the output dataset. For all options, other than `None` (which means no shuffling), the partitions of the underlying dataset/ddf will be randomly ordered. If `PER_PARTITION` is specified, each worker/process will also shuffle the rows within each partition before splitting and appending the data to a number (`out_files_per_proc`) of output files. Output files are distinctly mapped to each worker process. If `PER_WORKER` is specified, each worker will follow the same procedure as `PER_PARTITION`, but will re-shuffle each file after all data is persisted. This results in a full shuffle of the data processed by each worker. To improve performance, this option currently uses host-memory `BytesIO` objects for the intermediate persist stage. The `FULL` option is not yet implemented. partition_on : str or list(str) Columns to use for hive-partitioning. If this option is used, `preserve_files`, `output_files`, and `out_files_per_proc` cannot be specified, and `method` will be ignored. Also, the `PER_WORKER` shuffle will not be supported. preserve_files : bool Whether to preserve the original file-to-partition mapping of the base dataset. This option requires `method="subgraph"`, and is only available if the base dataset is known, and if it corresponds to csv or parquet format. If True, the `out_files_per_proc` option will be ignored. Default is False. output_files : dict, list or int The total number of desired output files. This option requires `method="subgraph"`. When `out_files_per_proc=None`, the default is the number of underlying Dask partitions. When `out_files_per_proc` is set to an integer, the default is the product of that integer and the total number of workers in the Dask cluster. For further output-file control, this argument may also be used to pass a dictionary mapping the output file names to partition indices, or a list of desired output-file names. out_files_per_proc : integer Number of output files that each process will use to shuffle an input partition. Default is 1. If `method="worker"`, the total number of output files will always be the total number of Dask workers, multiplied by this argument. If `method="subgraph"`, the total number of files is determined by `output_files` (and `out_files_per_proc` must be 1 if a dictionary is specified). row_group_size : integer Maximum number of rows to include in each Parquet row-group. By default, the maximum row-group size will be chosen by the backend Parquet engine (cudf or pyarrow). Note that cudf currently prohibits this value from being less than `5000` rows. If smaller row-groups are necessary, try calling `to_cpu()` before writing to disk. num_threads : integer Number of IO threads to use for writing the output dataset. For `0` (default), no dedicated IO threads will be used. dtypes : dict Dictionary containing desired datatypes for output columns. Keys are column names, values are datatypes. suffix : str or False File-name extension to use for all output files. This argument is ignored if a specific list of file names is specified using the ``output_files`` option. If ``preserve_files=True``, this suffix will be appended to the original name of each file, unless the original extension is ".csv", ".parquet", ".avro", or ".orc" (in which case the old extension will be replaced). cats : list of str, optional List of categorical columns conts : list of str, optional List of continuous columns labels : list of str, optional List of label columns method : {"subgraph", "worker"} General algorithm to use for the parallel graph execution. In order to minimize memory pressure, `to_parquet` will use a `"subgraph"` by default. This means that we segment the full Dask task graph into a distinct subgraph for each output file (or output-file group). Then, each of these subgraphs is executed, in full, by the same worker (as a single large task). In some cases, it may be more ideal to prioritize concurrency. In that case, a worker-based approach can be used by specifying `method="worker"`. write_hugectr_keyset : bool, optional Whether to write a HugeCTR keyset output file ("_hugectr.keyset"). Writing this file can be very slow, and should only be done if you are planning to ingest the output data with HugeCTR. Default is False. """ preserve_partitions = False if partition_on: # Check that the user is not expecting a specific output-file # count/structure that is not supported if output_files: raise ValueError("`output_files` not supported when `partition_on` is used.") if out_files_per_proc: raise ValueError("`out_files_per_proc` not supported when `partition_on` is used.") if preserve_files: raise ValueError("`preserve_files` not supported when `partition_on` is used.") else: # Check that method (algorithm) is valid if method not in ("subgraph", "worker"): raise ValueError(f"{method} not a recognized method for `Dataset.to_parquet`") # Deal with method-specific defaults if method == "worker": if output_files or preserve_files: raise ValueError("output_files and preserve_files require `method='subgraph'`") output_files = False elif preserve_files and output_files: raise ValueError("Cannot specify both preserve_files and output_files.") elif not (output_files or preserve_files): if out_files_per_proc: # Default "subgraph" behavior - Set output_files to the # total umber of workers, multiplied by out_files_per_proc try: nworkers = len(global_dask_client().cluster.workers) except AttributeError: nworkers = 1 output_files = nworkers * out_files_per_proc else: # Preserve original Dask partitions if output_files, # preserve_files AND out_files_per_proc are all None preserve_partitions = True # Replace None/False suffix argument with "" suffix = suffix or "" # Check shuffle argument shuffle = _check_shuffle_arg(shuffle) if isinstance(output_files, dict) or (not output_files and preserve_files): # Do not shuffle partitions if we are preserving files or # if a specific file-partition mapping is already specified ddf = self.to_ddf() else: ddf = self.to_ddf(shuffle=shuffle) # Check if partitions should be preserved if preserve_partitions: output_files = ddf.npartitions # Deal with `method=="subgraph"`. # Convert `output_files` argument to a dict mapping if output_files: # NOTES on `output_files`: # # - If a list of file names is specified, a contiguous range of # output partitions will be mapped to each file. The same # procedure is used if an integer is specified, but the file # names will be written as "part_*". # # - When `output_files` is used, the `output_files_per_proc` # argument will be interpreted as the desired number of output # files to write within the same task at run time (enabling # input partitions to be shuffled into multiple output files). # # - Passing a list or integer to `output_files` will preserve # the original ordering of the input data as long as # `out_files_per_proc` is set to `1` (or `None`), and # `shuffle==None`. # # - If a dictionary is specified, excluded partition indices # will not be written to disk. # # - To map multiple output files to a range of input partitions, # dictionary-input keys should correspond to a tuple of file # names. # Use out_files_per_proc to calculate how # many output files should be written within the # same subgraph. Note that we must a files_per_task = out_files_per_proc or 1 required_npartitions = ddf.npartitions if isinstance(output_files, int): required_npartitions = output_files files_per_task = min(files_per_task, output_files) elif isinstance(output_files, list): required_npartitions = len(output_files) files_per_task = min(files_per_task, len(output_files)) elif out_files_per_proc: raise ValueError( "Cannot specify out_files_per_proc if output_files is " "defined as a dictionary mapping. Please define each " "key in output_files as a tuple of file names if you " "wish to have those files written by the same process." ) # Repartition ddf if necessary if ddf.npartitions < required_npartitions: ddf = ddf.clear_divisions().repartition(npartitions=required_npartitions) # Construct an output_files dictionary if necessary if isinstance(output_files, int): output_files = [f"part_{i}" + suffix for i in range(output_files)] if isinstance(output_files, list): new = {} file_count = 0 split = math.ceil(ddf.npartitions / len(output_files)) for i in range(0, len(output_files), files_per_task): fns = output_files[i : i + files_per_task] start = i * split stop = min(start + split * len(fns), ddf.npartitions) if start < stop: new[tuple(fns)] = np.arange(start, stop) file_count += len(fns) # let user know they will not have expected number of output files. if file_count < len(output_files): warnings.warn( f"Only creating {file_count} files. Did not have enough " f"partitions to create {len(output_files)} files." ) output_files = new suffix = "" # Don't add a suffix later - Names already include it if not isinstance(output_files, dict): raise TypeError(f"{type(output_files)} not a supported type for `output_files`.") # If we are preserving files, use the stored dictionary, # or use file_partition_map to extract the mapping elif preserve_files: try: _output_files = self.base_dataset.file_partition_map except AttributeError as e: raise AttributeError( f"`to_parquet(..., preserve_files=True)` is not currently supported " f"for datasets with a {type(self.base_dataset.engine)} engine. Check " f"that `dataset.base_dataset` is backed by csv or parquet files." ) from e if suffix == "": output_files = _output_files else: output_files = {} for fn, rgs in _output_files.items(): split_fn = fn.split(".") if split_fn[-1] in ("parquet", "avro", "orc", "csv"): output_files[".".join(split_fn[:-1]) + suffix] = rgs else: output_files[fn + suffix] = rgs suffix = "" # Don't add a suffix later - Names already include it schema = self.schema.copy() if dtypes: _meta = _set_dtypes(ddf._meta, dtypes) ddf = ddf.map_partitions(_set_dtypes, dtypes, meta=_meta) for col_name, col_dtype in dtypes.items(): schema[col_name] = schema[col_name].with_dtype(col_dtype) fs = get_fs_token_paths(output_path)[0] fs.mkdirs(str(output_path), exist_ok=True) tf_metadata = TensorflowMetadata.from_merlin_schema(schema) tf_metadata.to_proto_text_file(output_path) metadata_path = fs.sep.join([str(output_path), MERLIN_METADATA_DIR_NAME]) fs.mkdirs(metadata_path, exist_ok=True) tf_metadata.to_json_file(metadata_path) # Output dask_cudf DataFrame to dataset _ddf_to_dataset( ddf, fs, output_path, shuffle, output_files, out_files_per_proc, cats or [], conts or [], labels or [], "parquet", num_threads, self.cpu, suffix=suffix, row_group_size=row_group_size, partition_on=partition_on, schema=schema if write_hugectr_keyset else None, )
[docs] def to_hugectr( self, output_path, cats, conts, labels, shuffle=None, file_partition_map=None, out_files_per_proc=None, num_threads=0, dtypes=None, ): """Writes out to a hugectr dataset Parameters ---------- output_path : string Path to write processed/shuffled output data cats : list of str List of categorical columns conts : list of str List of continuous columns labels : list of str List of label columns shuffle : merlin.io.Shuffle, optional How to shuffle the output dataset. Shuffling is only performed if the data is written to disk. For all options, other than `None` (which means no shuffling), the partitions of the underlying dataset/ddf will be randomly ordered. If `PER_PARTITION` is specified, each worker/process will also shuffle the rows within each partition before splitting and appending the data to a number (`out_files_per_proc`) of output files. Output files are distinctly mapped to each worker process. If `PER_WORKER` is specified, each worker will follow the same procedure as `PER_PARTITION`, but will re-shuffle each file after all data is persisted. This results in a full shuffle of the data processed by each worker. To improve performance, this option currently uses host-memory `BytesIO` objects for the intermediate persist stage. The `FULL` option is not yet implemented. file_partition_map : dict Dictionary mapping of output file names to partition indices that should be written to that file name. If this argument is passed, only the partitions included in the dictionary will be written to disk, and the `output_files_per_proc` argument will be ignored. out_files_per_proc : integer Number of files to create (per process) after shuffling the data num_threads : integer Number of IO threads to use for writing the output dataset. For `0` (default), no dedicated IO threads will be used. dtypes : dict Dictionary containing desired datatypes for output columns. Keys are column names, values are datatypes. """ # For now, we must move to the GPU to # write an output dataset. # TODO: Support CPU-mode output self.to_gpu() shuffle = _check_shuffle_arg(shuffle) ddf = self.to_ddf(shuffle=shuffle) if dtypes: _meta = _set_dtypes(ddf._meta, dtypes) ddf = ddf.map_partitions(_set_dtypes, dtypes, meta=_meta) fs = get_fs_token_paths(output_path)[0] fs.mkdirs(output_path, exist_ok=True) self.schema.write(output_path) # Output dask_cudf DataFrame to dataset, _ddf_to_dataset( ddf, fs, output_path, shuffle, file_partition_map, out_files_per_proc, cats, conts, labels, "hugectr", num_threads, self.cpu, schema=self.schema, )
[docs] def to_npy( self, output_file: str, append: bool = False, ): """Converts a dataset into an npy file, can append if data is larger than memory Parameters ---------- output_file : str The output file path for the resulting npy file append : bool, optional Enables append mode for larger that memory data, by default False """ data = self.to_ddf() if append: data = Dataset(data) itr = iter(data.to_iter()) with NpyAppendArray(output_file) as nf: for df in itr: to_write = dataframe_columnwise_explode(df) # after the explode there may not be object series anymore if "object" in to_write.dtypes.values and append: raise TypeError("Cannot append object columns") if (to_write.isnull()).any().any(): raise ValueError("Cannot convert data because null values were detected") nf.append(to_write.to_numpy()) else: to_write = dataframe_columnwise_explode(data.compute()) if "object" in to_write.dtypes.values and append: raise TypeError("Cannot append object columns") if (to_write.isnull()).any().any(): raise ValueError("Cannot convert data because null values were detected") np.save(output_file, to_write.to_numpy())
@property def num_rows(self): return self.engine.num_rows @property def npartitions(self): return self.to_ddf().npartitions
[docs] def validate_dataset(self, **kwargs): """Validate for efficient processing. The purpose of this method is to validate that the Dataset object meets the minimal requirements for efficient NVTabular processing. For now, this criteria requires the data to be in parquet format. Example Usage:: dataset = Dataset("/path/to/data_pq", engine="parquet") assert validate_dataset(dataset) Parameters ----------- **kwargs : Key-word arguments to pass down to the engine's validate_dataset method. For the recommended parquet format, these arguments include `add_metadata_file`, `row_group_max_size`, `file_min_size`, and `require_metadata_file`. For more information, see `ParquetDatasetEngine.validate_dataset`. Returns ------- valid : bool `True` if the input dataset is valid for efficient NVTabular processing. """ # Check that the dataset format is Parquet if not isinstance(self.engine, ParquetDatasetEngine): msg = ( "NVTabular is optimized for the parquet format. Please use " "the to_parquet method to convert your dataset." ) warnings.warn(msg) return False # Early return return self.engine.validate_dataset(**kwargs)
[docs] def regenerate_dataset( self, output_path, columns=None, output_format="parquet", compute=True, **kwargs, ): """EXPERIMENTAL: Regenerate an NVTabular Dataset for efficient processing by writing out new Parquet files. In contrast to default ``to_parquet`` behavior, this method preserves the original ordering. Example Usage:: dataset = Dataset("/path/to/data_pq", engine="parquet") dataset.regenerate_dataset( out_path, part_size="1MiB", file_size="10MiB" ) Parameters ----------- output_path : string Root directory path to use for the new (regenerated) dataset. columns : list(string), optional Subset of columns to include in the regenerated dataset. output_format : string, optional Format to use for regenerated dataset. Only "parquet" (default) is currently supported. compute : bool, optional Whether to compute the task graph or to return a Delayed object. By default, the graph will be executed. **kwargs : Key-word arguments to pass down to the engine's regenerate_dataset method. See `ParquetDatasetEngine.regenerate_dataset` for more information. Returns ------- result : int or Delayed If `compute=True` (default), the return value will be an integer corresponding to the number of generated data files. If `False`, the returned value will be a `Delayed` object. """ # Check that the desired output format is Parquet if output_format not in ["parquet"]: msg = ( f"NVTabular is optimized for the parquet format. " f"{output_format} is not yet a supported output format for " f"regenerate_dataset." ) raise ValueError(msg) result = ParquetDatasetEngine.regenerate_dataset(self, output_path, columns=None, **kwargs) if compute: return result.compute() else: return result
[docs] def infer_schema(self, n=1): """Create a schema containing the column names and inferred dtypes of the Dataset Args: n (int, optional): Number of rows to sample to infer the dtypes. Defaults to 1. """ dtypes = {} dtypes = self.sample_dtypes(n=n, annotate_lists=True) column_schemas = [] for column, dtype_info in dtypes.items(): dtype_val = dtype_info["dtype"] dims = DefaultShapes.LIST if dtype_info["is_list"] else DefaultShapes.SCALAR col_schema = ColumnSchema(column, dtype=dtype_val, dims=dims) column_schemas.append(col_schema) self.schema = Schema(column_schemas) return self.schema
[docs] def sample_dtypes(self, n=1, annotate_lists=False): """Return the real dtypes of the Dataset Use cached metadata if this operation was already performed. Otherwise, call down to the underlying engine for sampling logic. """ if self._real_meta.get(n, None) is None: _real_meta = self.engine.sample_data(n=n) if self.dtypes: _real_meta = _set_dtypes(_real_meta, self.dtypes) self._real_meta[n] = _real_meta if annotate_lists: _real_meta = self._real_meta[n] annotated = {} for col in _real_meta.columns: is_list = is_list_dtype(_real_meta[col]) dtype = list_val_dtype(_real_meta[col]) if is_list else _real_meta[col].dtype annotated[col] = {"dtype": dtype, "is_list": is_list} return annotated return self._real_meta[n].dtypes
@classmethod def _bind_dd_method(cls, name): """Bind Dask-Dataframe method to the Dataset class""" def meth(self, *args, **kwargs): _meth = getattr(self.to_ddf(), name) return _meth(*args, **kwargs) meth.__name__ = name setattr(cls, name, meth)
# Bind (simple) Dask-Dataframe Methods for op in ["compute", "persist", "head", "tail"]: Dataset._bind_dd_method(op) def _set_dtypes(chunk, dtypes): for col, dtype in dtypes.items(): if isinstance(dtype, str) and ("hex" in dtype): chunk[col] = hex_to_int(chunk[col]) else: chunk[col] = chunk[col].astype(dtype) return chunk