This is an automated pull request to update the first-party submodule for pytorch/FBGEMM.

New submodule commit: https://github.com/pytorch/FBGEMM/commit/f99e1616630e3b78e3bccd0ceb9e50e8e82409f1

Test Plan: Ensure that CI jobs succeed on GitHub before landing.

This is an automated pull request to update the first-party submodule for pytorch/FBGEMM.

New submodule commit: https://github.com/pytorch/FBGEMM/commit/e8d19d8ed9920d8e3a53c3baadb6cb239798646d

Test Plan: Ensure that CI jobs succeed on GitHub before landing.

This is an automated pull request to update the first-party submodule for pytorch/FBGEMM.

New submodule commit: https://github.com/pytorch/FBGEMM/commit/61de99ff78ae0c5b5b97c5d0c091e85c23cba453

Test Plan: Ensure that CI jobs succeed on GitHub before landing.

Summary: We expand kjt interface to include these new calls: sdd_labels -> names of tensors to transmit sdd_splits -> which is shape of internal tensors, split by key sdd_tensors -> the actual tensor data to transmit sdd_init -> builds a new kjt from raw colllective output

Next we changed the KJT a2a to have awaitables. The first awaitable is to transmit the tensor splits so each rank will know the size of the tensors it will receive (this collective call is blocking). The second awaitable is asynchronous and it transmits the tensor data to the correct ranks.

Finally we allowed input dist fusion by combining the first awaitable in the KJT a2a for all KJTs to execute one after another synchronously, and then calling the second awaitable to transmit the actual tensor data asynchronously.

TODO: Support gloo

Differential Revision: D39520093

LaMa Project: L1138451

Summary: -> this diff discards following sharding combination: (RW/CW/TW/TWRW/TWCW) with batched_dense.

Also, pruning is being removed from proposers.py as it is no longer needed.

Differential Revision: D36644029

This is an automated pull request to update the first-party submodule for pytorch/FBGEMM.

New submodule commit: https://github.com/pytorch/FBGEMM/commit/41d598cd50700fa9e85f596d6c230205632c025d

Test Plan: Ensure that CI jobs succeed on GitHub before landing.

This is an automated pull request to update the first-party submodule for pytorch/FBGEMM.

New submodule commit: https://github.com/pytorch/FBGEMM/commit/5c804646571b13afd860a74897c8c99ca7c4e1b9

Test Plan: Ensure that CI jobs succeed on GitHub before landing.

Hi team, I'm trying to use torchrec-nightly with torch 1.12 and CUDA 11.2. But when I import torchrec, I get the following:

>>> import torchrec
File fbgemm_gpu_py.so not found

A similar issue was reported on the DLRM issue tracker https://github.com/facebookresearch/dlrm/issues/256. Any ideas?

From README

Description

This shows a prototype of integrating a TorchRec based training loop utilizing TorchArrow's on-the-fly preprocessing. The main motivation is to show the utilization of TorchArrow's specialized domain UDFs. Here we use bucketize, firstx, as well as sigrid_hash to do some last-mile preprocessing over the criteo dataset in parquet format.

These three UDFs are extensively used in Meta's RecSys preprocessing stack. Notably, these UDFs can be used to easily adjust the proprocessing script to any model changes. For example, if we wish to change the size of our embedding tables, without sigrid_hash, we would need to rerun a bulk offline preproc to ensure that all indicies are within bounds. Bucketize lets us easily convert dense features into sparse features, with flexibility of what the bucket borders are. firstx lets us easily prune sparse ids (note, that this doesn't provide any functional features, but is in the preproc script as demonstration).

Installations and Usage

Download the criteo tsv files (see the README in the main DLRM example). Use the nvtabular script (in torchrec/datasets/scripts/nvt/) to convert the TSV files to parquet.

Install torcharrow from https://github.com/facebookresearch/torcharrow1`

pip install torchdata

Usage

torchx run -s local_cwd dist.ddp -j 1x4 --script examples/torcharrow/run.py -- --parquet_directory /home/criteo_parquet

The preprocessing logic is in dataloader.py

Extentions/Future work

• We will eventually integrate with the up and coming DataLoader2, which will allow us to utilize a prebuilt solution to collate our dataframe batches to dense tensors, or TorchRec's KeyedJaggedTensors (rather than doing this by hand).
• Building an easier solution/more performant to convert parquet -> IterableDataPipe[DataFrame]
• Some functional abilities are not yet available (such as make_named_row, etc).
• More RecSys UDFs to come! Please let us know if you have any suggestions.

This is an automated pull request to update the first-party submodule for pytorch/FBGEMM.

New submodule commit: https://github.com/pytorch/FBGEMM/commit/243010f7c3a6887e102add293cb5a530aa884240

Test Plan: Ensure that CI jobs succeed on GitHub before landing.

This is an automated pull request to update the first-party submodule for pytorch/FBGEMM.

New submodule commit: https://github.com/pytorch/FBGEMM/commit/6b629495e0ac2f5ab8fd7472fb40d19194077915

Test Plan: Ensure that CI jobs succeed on GitHub before landing.

Summary: Currently the apply_optimizer_in_backward API lives in both torchrec and distributed folders. We want to migrate all usage of torchrec's apply_optimizer_in_backward to distributed's to unify the API.

Reviewed By: colin2328, YLGH

Differential Revision: D42319179

Summary: The current implementation computes each metric state individually. This diff vectorizes these operations. Some key changes in this diff:

• states vectorization in all supported metrics
• In auc metric, stores its states with tensor instead of a list of one element. This is to make the states vectorization logic in all metrics be consistent.

Most parts of this diff are the same as D40419238 (https://github.com/pytorch/torchrec/commit/50c861a4debb6d0d8bd55ddb27452e89f2d19d51). The main difference is that this one has added the backward compatibility support for old model checkpoints.

Differential Revision: D42161727

Summary: Currently in storage_reservations.py, we use percentage to only reserve hbm memory, no ddr memory (https://fburl.com/code/166evbil). This diff makes stats.py behave accordingly.

Reviewed By: joshuadeng

Differential Revision: D42329036

Hello, I am trying to pre-process criteo TB dataset using NVTabular. I run the nvt_preproc.sh nvt_preproc but get errors.

This is the error code.

2022-12-21 15:04:51,981 - distributed.preloading - INFO - Creating preload: dask_cuda.initialize
2022-12-21 15:04:51,982 - distributed.preloading - INFO - Import preload module: dask_cuda.initialize
/N/u/haofeng/BigRed200/anaconda3/envs/rapids-22.12/lib/python3.9/site-packages/merlin/io/dataset.py:251: UserWarning: Initializing an NVTabular Dataset in CPU mode.This is an experimental feature with extremely limited support!
  warnings.warn(
2022-12-21 16:25:17,116 - distributed.worker - WARNING - Compute Failed
Key:       ('read-csv-90ef43a055712c364b500b995ea7e3d5', 0)
Function:  execute_task
args:      ((subgraph_callable-12d6d57f-182a-4e3a-b65c-e4ee508dcef9, [(<function read_block_from_file at 0x7ff756010a60>, <OpenFile '/N/scratch/haofeng/TB/raw/day_0'>, 0, 128272716, b'\n'), None, True, False]))
kwargs:    {}
Exception: 'ValueError("Mismatched dtypes found in `pd.read_csv`/`pd.read_table`.\\n\\n+--------+---------+----------+\\n| Column | Found   | Expected |\\n+--------+---------+----------+\\n| int_1  | float64 | int64    |\\n| int_11 | float64 | int64    |\\n| int_6  | float64 | int64    |\\n+--------+---------+----------+\\n\\nUsually this is due to dask\'s dtype inference failing, and\\n*may* be fixed by specifying dtypes manually by adding:\\n\\ndtype={\'int_1\': \'float64\',\\n       \'int_11\': \'float64\',\\n       \'int_6\': \'float64\'}\\n\\nto the call to `read_csv`/`read_table`.\\n\\nAlternatively, provide `assume_missing=True` to interpret\\nall unspecified integer columns as floats.")'

finished splitting the last day, took 4719.522433280945
handling the input paths: ['/N/scratch/haofeng/TB/raw/day_0', '/N/scratch/haofeng/TB/raw/day_1', '/N/scratch/haofeng/TB/raw/day_2', '/N/scratch/haofeng/TB/raw/day_3', '/N/scratch/haofeng/TB/raw/day_4', '/N/scratch/haofeng/TB/raw/day_5', '/N/scratch/haofeng/TB/raw/day_6', '/N/scratch/haofeng/TB/raw/day_7', '/N/scratch/haofeng/TB/raw/day_8', '/N/scratch/haofeng/TB/raw/day_9', '/N/scratch/haofeng/TB/raw/day_10', '/N/scratch/haofeng/TB/raw/day_11', '/N/scratch/haofeng/TB/raw/day_12', '/N/scratch/haofeng/TB/raw/day_13', '/N/scratch/haofeng/TB/raw/day_14', '/N/scratch/haofeng/TB/raw/day_15', '/N/scratch/haofeng/TB/raw/day_16', '/N/scratch/haofeng/TB/raw/day_17', '/N/scratch/haofeng/TB/raw/day_18', '/N/scratch/haofeng/TB/raw/day_19', '/N/scratch/haofeng/TB/raw/day_20', '/N/scratch/haofeng/TB/raw/day_21', '/N/scratch/haofeng/TB/raw/day_22', '/N/scratch/haofeng/TB/raw/day_23.part0', '/N/scratch/haofeng/TB/raw/day_23.part1']
Traceback (most recent call last):
  File "/geode2/home/u030/haofeng/BigRed200/torchrec/torchrec/datasets/scripts/nvt/convert_tsv_to_parquet.py", line 107, in <module>
    convert_tsv_to_parquet(args.input_path, args.output_base_path)
  File "/geode2/home/u030/haofeng/BigRed200/torchrec/torchrec/datasets/scripts/nvt/convert_tsv_to_parquet.py", line 72, in convert_tsv_to_parquet
    tsv_dataset = nvt.Dataset(input_paths, **config)
  File "/N/u/haofeng/BigRed200/anaconda3/envs/rapids-22.12/lib/python3.9/site-packages/merlin/io/dataset.py", line 346, in __init__
    self.infer_schema()
  File "/N/u/haofeng/BigRed200/anaconda3/envs/rapids-22.12/lib/python3.9/site-packages/merlin/io/dataset.py", line 1127, in infer_schema
    dtypes = self.sample_dtypes(n=n, annotate_lists=True)
  File "/N/u/haofeng/BigRed200/anaconda3/envs/rapids-22.12/lib/python3.9/site-packages/merlin/io/dataset.py", line 1147, in sample_dtypes
    _real_meta = self.engine.sample_data(n=n)
  File "/N/u/haofeng/BigRed200/anaconda3/envs/rapids-22.12/lib/python3.9/site-packages/merlin/io/dataset_engine.py", line 71, in sample_data
    _head = _ddf.partitions[partition_index].head(n)
  File "/N/u/haofeng/BigRed200/anaconda3/envs/rapids-22.12/lib/python3.9/site-packages/dask/dataframe/core.py", line 1265, in head
    return self._head(n=n, npartitions=npartitions, compute=compute, safe=safe)
  File "/N/u/haofeng/BigRed200/anaconda3/envs/rapids-22.12/lib/python3.9/site-packages/dask/dataframe/core.py", line 1299, in _head
    result = result.compute()
  File "/N/u/haofeng/BigRed200/anaconda3/envs/rapids-22.12/lib/python3.9/site-packages/dask/base.py", line 315, in compute
    (result,) = compute(self, traverse=False, **kwargs)
  File "/N/u/haofeng/BigRed200/anaconda3/envs/rapids-22.12/lib/python3.9/site-packages/dask/base.py", line 600, in compute
    results = schedule(dsk, keys, **kwargs)
  File "/N/u/haofeng/BigRed200/anaconda3/envs/rapids-22.12/lib/python3.9/site-packages/distributed/client.py", line 3122, in get
    results = self.gather(packed, asynchronous=asynchronous, direct=direct)
  File "/N/u/haofeng/BigRed200/anaconda3/envs/rapids-22.12/lib/python3.9/site-packages/distributed/client.py", line 2291, in gather
    return self.sync(
  File "/N/u/haofeng/BigRed200/anaconda3/envs/rapids-22.12/lib/python3.9/site-packages/distributed/utils.py", line 339, in sync
    return sync(
  File "/N/u/haofeng/BigRed200/anaconda3/envs/rapids-22.12/lib/python3.9/site-packages/distributed/utils.py", line 406, in sync
    raise exc.with_traceback(tb)
  File "/N/u/haofeng/BigRed200/anaconda3/envs/rapids-22.12/lib/python3.9/site-packages/distributed/utils.py", line 379, in f
    result = yield future
  File "/N/u/haofeng/BigRed200/anaconda3/envs/rapids-22.12/lib/python3.9/site-packages/tornado/gen.py", line 762, in run
    value = future.result()
  File "/N/u/haofeng/BigRed200/anaconda3/envs/rapids-22.12/lib/python3.9/site-packages/distributed/client.py", line 2154, in _gather
    raise exception.with_traceback(traceback)
  File "/N/u/haofeng/BigRed200/anaconda3/envs/rapids-22.12/lib/python3.9/site-packages/dask/optimization.py", line 990, in __call__
    return core.get(self.dsk, self.outkey, dict(zip(self.inkeys, args)))
  File "/N/u/haofeng/BigRed200/anaconda3/envs/rapids-22.12/lib/python3.9/site-packages/dask/core.py", line 149, in get
    result = _execute_task(task, cache)
  File "/N/u/haofeng/BigRed200/anaconda3/envs/rapids-22.12/lib/python3.9/site-packages/dask/core.py", line 119, in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
  File "/N/u/haofeng/BigRed200/anaconda3/envs/rapids-22.12/lib/python3.9/site-packages/dask/dataframe/io/csv.py", line 141, in __call__
    df = pandas_read_text(
  File "/N/u/haofeng/BigRed200/anaconda3/envs/rapids-22.12/lib/python3.9/site-packages/dask/dataframe/io/csv.py", line 196, in pandas_read_text
    coerce_dtypes(df, dtypes)
  File "/N/u/haofeng/BigRed200/anaconda3/envs/rapids-22.12/lib/python3.9/site-packages/dask/dataframe/io/csv.py", line 297, in coerce_dtypes
    raise ValueError(msg)
ValueError: Mismatched dtypes found in `pd.read_csv`/`pd.read_table`.

+--------+---------+----------+
| Column | Found   | Expected |
+--------+---------+----------+
| int_1  | float64 | int64    |
| int_11 | float64 | int64    |
| int_6  | float64 | int64    |
+--------+---------+----------+

Usually this is due to dask's dtype inference failing, and
*may* be fixed by specifying dtypes manually by adding:

dtype={'int_1': 'float64',
       'int_11': 'float64',
       'int_6': 'float64'}

to the call to `read_csv`/`read_table`.

Alternatively, provide `assume_missing=True` to interpret
all unspecified integer columns as floats.
*** Error in `/N/u/haofeng/BigRed200/anaconda3/envs/rapids-22.12/bin/python': corrupted size vs. prev_size: 0x00007ff06c000b00 ***

I read the error code, according to my current understanding, I think it is caused by a data type mismatch when reading the file. But I do download the original files from this website original data. I'm hoping to get some help telling me why this is and how I should fix it.

I still have two more questions, the first one is that after I downloaded these files and decompressed them with gzip -dk $filename, I got some files which names are day_0, day_1 ... instead of day_0.csv, day_1.csv. Will this cause some mistakes? I think this is only a difference in the suffix, but I am not sure.
Another question is it seems that I am running NVTabular Dataset in CPU mode, I have already install cupy, rmm and some other libraries, what should I do to run this program in CUDA mode?
Thank you for your help!