Repository: NVIDIA-Merlin/NVTabular (language: python)

Issue:
[FEA] Sequential / Session-based recommendation and time series support - Group by sorting values by timestamp
**Is your feature request related to a problem? Please describe.**
In order to support **session-based / sequence recommendation** and also **time series use cases**, we need NVTabular to be able to group data by some columns (e.g. user id, session id), sorted by another column (e.g. usually timestamp), and aggregate other columns with these different aggregation functions that takes order into account: '**list**', '**first**', '**last**'

**Describe the solution you'd like**
```python
groupby_features = ['user_id', 'session_id', 'timestamp', 'prod_id', 'prod_categ'] >> ops.Groupby(
    groupby_cols=['user_id', 'session_id'], sort_column=['timestamp'], 
    aggregated_cols={'prod_id': 'list',
                   'prod_categ': 'list',
                   'timestamp': ['first', 'last']}
)
processor = nvtabular.Workflow(groupby_features)
```

**Describe alternatives you've considered**
It might be ok is some NVT ops do not support the resulting list columns, like `Categorify` and `Standardize`, because those ops can be done before the grouping.
But the `LambdaOp` should support list columns to allow, for example, to extract the length of the lists or truncate the lists.

**Additional context**
This feature can be accomplished by different data frame frameworks:

***Pandas***
In this example, the preliminary sorting of the rows is respected by the group by, so that the aggregated columns ('item_id') will be sorted by timestamp.

```python
interactions_df.sort_values('timestamp', inplace=True)
sessions_df = interactions_df.groupby('session_id').agg('item_id': 'list')
```
P.s. `cudf / dask_cudf` also supports the 'list' aggregation function. But it does not guarantee that the data frame ordering will be respected, as pandas does.

***PySpark***
```python
import pyspark.sql.functions as F

# This window ensures the sorting of aggregated columns by another column: timestamp
session_window = Window.partitionBy('user_session').orderBy('timestamp')

df.select('user_session',                        
                        F.collect_list('product_idx').over(session_window).alias('product_seq'), 
                        F.first('event_time_ts').over(session_window).alias('session_start_ts'),
                        F.last('event_time_ts').over(session_window).alias('session_end_ts'),
                        ) \
                .groupBy('user_session').agg(
                    # The F.max() is necessary for the list column because F.collect_list() will create an incremental
                    # list of values for each value, sorted by time, and F.max() returns the longest list (containing all items of the sequence).
                    F.max('product_seq').alias('product_seq'),
                    F.min('session_start_ts').alias('session_start_ts'),
                    F.max('session_end_ts').alias('session_end_ts'),
                 )
```
This issue was extracted from #355 , which is broader in scope, so that it is implemented independently.
Other related issues: #92 #325 
Handle hive-partitioning in NVTabular.dataset.Dataset
Closes #642 
Addresses global shuffle component of #641 

Thu purpose of this PR is to improve handling of hive-partitioned parquet data in NVTabular.  Since the Dataset API already uses `dask.dataframe.read_parquet`, there is currenlty no "correcness" issue with **reading** hive-partitioned data.  However, (1) there is no convenient mechanism to write hive-partitioned data, and (2) the read stage typically results in many small partitions (rather than a single partition for each input directory).

- **Solution to (1)**:  The `Dataset.to_parquet` method now supports a `partition_on=` argument.  This is designed to match the same option in `dask.dataframe`/`dask_cudf`.  If the user passes a list of 1+ columns with this argument, the output data will be shuffled at IO time into a distinct directory for each unique combination of those `partition_on` column values.  When multiple columns are use for partitioning (e.g. `["month", "day"]`), the directory structure is nested (so that the full path for an output file will look something like `"/month=Mar/day=30/part.0.parquet"`).
- **Solution to (2)**:  Since #641 will need a mechanism to ensure a unique mapping between specified column groups and ddf partitions, this PR adds a `Dataset.partition_by_keys` method to perform a global shuffle on the specified column group (`keys`) and return a new (shuffled) Dataset.  For general Dataset objects, this method will simply call `ddf.shuffle()` under the hood.  For Dataset objects that are backed by hive-partitioned data, however, we use the metadata stored in the file paths to avoid a full shuffle. In the future, this optimization can be pushed even further by directly agregating all IO tasks within the same hive-partition.  However, I suspect that shuch an optimization should be implemented in `dask.dataframe`.

**Example Usage**
```python
import pandas as pd
import dask.dataframe as dd
import dask
import nvtabular as nvt

path = "fake.data"

# Create a sample ddf
ddf = dask.datasets.timeseries(
    start="2000-01-01",
    end="2000-01-03",
    freq="600s",
    partition_freq="6h",
    seed=42,
).reset_index()
ddf['timestamp'] = ddf['timestamp'].dt.round('D').dt.day

# Convert to a Datset and write out hive-partitioned data to disk
keys = ["timestamp", "name"]
nvt.Dataset(ddf).to_parquet(path, partition_on=keys)
```

This will produce a directory structure like:
```
$ find fake.data/ -type d -print
fake.data/
fake.data/timestamp=1
fake.data/timestamp=1/name=Alice
fake.data/timestamp=1/name=Frank
fake.data/timestamp=1/name=Victor
fake.data/timestamp=1/name=George
fake.data/timestamp=1/name=Quinn
fake.data/timestamp=1/name=Kevin
fake.data/timestamp=1/name=Ursula
...
```

Then, you can read the data back in with NVT, and ensure that the ddf partitions are shuffled by `keys`:
```python
ds = nvt.Dataset(path, engine="parquet").shuffle_by_keys(keys)
ds.to_ddf().compute()
```
```
      id         x         y timestamp    name
0    991 -0.750009 -0.587392         1   Alice
1   1022  0.866823 -0.682096         1   Alice
2    991  0.467775  0.683211         1   Alice
3    967  0.534984 -0.931405         1     Bob
4    991 -0.149716 -0.651939         1     Bob
..   ...       ...       ...       ...     ...
25   964  0.843602  0.598580         3  Yvonne
26   961  0.853070 -0.987596         3  Yvonne
27   947  0.934162  0.190069         3  Yvonne
28  1024 -0.107280  0.662606         3  Yvonne
29  1006  0.169090 -0.784889         3   Zelda

[288 rows x 5 columns]
```


<!--

Thank you for contributing to NVTabular :)

Here are some guidelines to help the review process go smoothly.

1. Please write a description in this text box of the changes that are being
   made.

2. Please ensure that you have written units tests for the changes made/features
   added.

3. If you are closing an issue please use one of the automatic closing words as
   noted here: https://help.github.com/articles/closing-issues-using-keywords/

4. If your pull request is not ready for review but you want to make use of the
   continuous integration testing facilities please label it with `[WIP]`.

5. If your pull request is ready to be reviewed without requiring additional
   work on top of it, then remove the `[WIP]` label (if present) and replace
   it with `[REVIEW]`. If assistance is required to complete the functionality,
   for example when the C/C++ code of a feature is complete but Python bindings
   are still required, then add the label `[HELP-REQ]` so that others can triage
   and assist. The additional changes then can be implemented on top of the
   same PR. If the assistance is done by members of the rapidsAI team, then no
   additional actions are required by the creator of the original PR for this,
   otherwise the original author of the PR needs to give permission to the
   person(s) assisting to commit to their personal fork of the project. If that
   doesn't happen then a new PR based on the code of the original PR can be
   opened by the person assisting, which then will be the PR that will be
   merged.

6. Once all work has been done and review has taken place please do not add
   features or make changes out of the scope of those requested by the reviewer
   (doing this just add delays as already reviewed code ends up having to be
   re-reviewed/it is hard to tell what is new etc!). Further, please do not
   rebase your branch on master/force push/rewrite history, doing any of these
   causes the context of any comments made by reviewers to be lost. If
   conflicts occur against master they should be resolved by merging master
   into the branch used for making the pull request.

Many thanks in advance for your cooperation!

-->



Discussion / hints:
This feature is essential in our time series workflows as well. In our core workflow, the core pseudocode looks like this:

```python
df = load_some_df()
df['primary_id'] = flatten_ids(['ids','to','combine'])
df.sort_values('time_id')
df.groupby('primary_id')
```
The changes proposed here would cover this case.
We also allow for per-group scaling down to the standard distribution. This would likely also be relevant to session based recommendation (scaling values by user or other key)



On another note, on the dataloading side, it will likely also be pertinent to allow for sliding-window access-time tensor building on the collected lists. As an example (ignoring NVT conventions to illustrate the point):

```python
list = [10,20,30]
session_loader = nvt.session_loader(list,window=2)
session_loader[0] #Result is [10,20]
session_loader[1] #Result is [20,30]

```
Thanks for writing up this issue @gabrielspmoreira - As you know, the **transform** `ops` available in NVTabular are all "partition-wise" operations (they do not require any shuffling of data between distinct dask.DataFrame partitions).  There are stat operators that **do** move data around, but the statistics/groupby results are always expected to be relatvely small, and the results are always persisted to disk at the end of the `fit` stage.  The `GroupbyJoin` operator, for example, is still a partition-wise transform, because the (relatively-small) groupby statistics are joined to the original data during the transform stage.

With the above in mind, I want to clarify: Is this issue requesting a **transform** operator that does not follow the "partition-wise" convention?  That is, is the goal to introduce a new type of transform that cannot be performed with a linear pass over the underlying ddf partitions or would this be a new "statistics" operation (only to be performed at "fit" time)?  In order to avoid a departure from the current/simple transform convention, it may actually make sense to separate this feature from the usual `Workflow`/`fit`/`tansform` pipeline altogether, and just introduce a `DataFrame.groupby` method.

cc @benfred (since I'm certainly interested in your thoughts here as well)
Thanks for your analysis @rjzamora. I understand that a GroupBy op requires full shuffle, because other rows from the same group (e.g. user id or session id) might not be in the same partition, right?

In this sense, I understand that it would be easier to implement `DataFrame.groupby` method outside the workflow. Therefore, it should be possible to do some custom sequence-wise transforms on the list columns after grouping. like: truncating the lists to a maximum length, counting the length of the sequence, getting the first value of the sequence. Such ops could be performed using a LambdaOp within our workflow, or our user could possibly implement then with dask_cudf `apply` outside the NVTabular workflow.

Another option would be having the GroupBy op as the first step in our pipeline, and then performing the subsequent ops (e.g. Categorify, LopOp, ...) on the list columns. 

During inference, the task of sequential recommendation and session-based recommendation is to provide a ranked list of items for a user or session, represented by a sequence of his past interactions. So, Triton should receive a batch with the last user interactions and return the recommendation list. The GroupBy op (ordered by the timestamp column), should be performed on inference the same way it was during preprocessing. So I understand that the GroupBy op should be within our workflow.
>Thanks for your analysis @rjzamora. I understand that a GroupBy op requires full shuffle, because other rows from the same group (e.g. user id or session id) might not be in the same partition, right?

Exactly right - I was using the term "partition-wise" to describe a transform that does not require a row/record to move between partitions, but I'm not sure of the best language here.

>Another option would be having the GroupBy op as the first step in our pipeline, and then performing the subsequent ops (e.g. Categorify, LopOp, ...) on the list columns.

The only reason I was suggesting a `Dataset.groupby` method (in lieue of a `Groupby` operator) is to make it impossible for users to try to use something like `ops.Groupby` in the "wrong" way.  If the grouping logic belongs to the Dataset object, and is designed to produce a **new** Dataset object, we are effectively requiring this operation to be the first step in a typical NVT pipeline.  In practice, users will get **much** better down-stream transform/dataloading performance if the result of the initial `Dataset.groupby` operation is persisted to disk (because it would relieve down-stream memory pressure), but an intermediate `to_parquet` call would not be a requirement.

  Note that my current thinking is a bit different (see below), but I am still thinking that `ops.Groupby` cannot move data between partitions.

>During inference, the task of sequential recommendation and session-based recommendation is to provide a ranked list of items for a user or session, represented by a sequence of his past interactions. So, Triton should receive a batch with the last user interactions and return the recommendation list. The GroupBy op (ordered by the timestamp column), should be performed on inference the same way it was during preprocessing. So I understand that the GroupBy op should be within our workflow.

Ah - This is a great point.  Then I agree that we ultimately **do need** something like `ops.Groupby`.  However, this operation will probably need to be a partition-wise transformation, and we will still need a new `Dataset.<something>` method to handle our data-shuffling needs for the training data.  Perhaps it would be sufficient to enable the user to shuffle the initial Dataset object by the groupby keys?  For example, if training/inference needs to group a batch of data by `['user_id', 'session_id’]`, this can be a partition-wise operation if the input data is guaranteed to have unique 'user_id’-‘session_id’ pairs within the same partition.  So, perhaps we add **both** a `Dataset.group_shuffle` method and `ops.Groupby`?  The only problem with this approach is that  the partitioning may change if the user persists the `Dataset.group_shuffle` result to disk, and then they try to read it back for processing.  We would need a new explicit option to partition by file in both the read and write (to_parquet).
cc @gabrielspmoreira @benfred 
<details><summary>Click to view CI Results</summary><pre>GitHub pull request #677 of commit 9fb2dd5d8fdc927f73405b5819e1ad46d6d9a4a3, no merge conflicts.
Running as SYSTEM
Setting status of 9fb2dd5d8fdc927f73405b5819e1ad46d6d9a4a3 to PENDING with url http://10.20.13.93:8080/job/nvtabular_tests/1989/ and message: 'Pending'
Using context: Jenkins Unit Test Run
Building in workspace /var/jenkins_home/workspace/nvtabular_tests
using credential ghub_token
Cloning the remote Git repository
Cloning repository https://github.com/NVIDIA/NVTabular.git
 > git init /var/jenkins_home/workspace/nvtabular_tests/nvtabular # timeout=10
Fetching upstream changes from https://github.com/NVIDIA/NVTabular.git
 > git --version # timeout=10
using GIT_ASKPASS to set credentials github token setup
 > git fetch --tags --force --progress -- https://github.com/NVIDIA/NVTabular.git +refs/heads/*:refs/remotes/origin/* # timeout=10
 > git config remote.origin.url https://github.com/NVIDIA/NVTabular.git # timeout=10
 > git config --add remote.origin.fetch +refs/heads/*:refs/remotes/origin/* # timeout=10
 > git config remote.origin.url https://github.com/NVIDIA/NVTabular.git # timeout=10
Fetching upstream changes from https://github.com/NVIDIA/NVTabular.git
using GIT_ASKPASS to set credentials github token setup
 > git fetch --tags --force --progress -- https://github.com/NVIDIA/NVTabular.git +refs/pull/677/*:refs/remotes/origin/pr/677/* # timeout=10
 > git rev-parse 9fb2dd5d8fdc927f73405b5819e1ad46d6d9a4a3^{commit} # timeout=10
Checking out Revision 9fb2dd5d8fdc927f73405b5819e1ad46d6d9a4a3 (detached)
 > git config core.sparsecheckout # timeout=10
 > git checkout -f 9fb2dd5d8fdc927f73405b5819e1ad46d6d9a4a3 # timeout=10
Commit message: "expand testing and fix bug"
 > git rev-list --no-walk 1f60ba950f935d104c7c8fa21742158698eba3eb # timeout=10
First time build. Skipping changelog.
[nvtabular_tests] $ /bin/bash /tmp/jenkins427843502843120314.sh
Obtaining file:///var/jenkins_home/workspace/nvtabular_tests/nvtabular
  Installing build dependencies: started
  Installing build dependencies: finished with status 'done'
  Getting requirements to build wheel: started
  Getting requirements to build wheel: finished with status 'done'
    Preparing wheel metadata: started
    Preparing wheel metadata: finished with status 'done'
Installing collected packages: nvtabular
  Running setup.py develop for nvtabular
Successfully installed nvtabular
All done! ✨ 🍰 ✨
93 files would be left unchanged.
/conda/envs/rapids/lib/python3.8/site-packages/isort/files.py:30: UserWarning: Likely recursive symlink detected to /var/jenkins_home/workspace/nvtabular_tests/nvtabular/images
  warn(f"Likely recursive symlink detected to {resolved_path}")
============================= test session starts ==============================
platform linux -- Python 3.8.8, pytest-6.2.2, py-1.10.0, pluggy-0.13.1
rootdir: /var/jenkins_home/workspace/nvtabular_tests/nvtabular, configfile: pyproject.toml
plugins: cov-2.11.1, xdist-2.2.1, forked-1.3.0
collected 763 items / 2 skipped / 761 selected

tests/unit/test_column_group.py .                                        [  0%]
tests/unit/test_column_similarity.py ......                              [  0%]
tests/unit/test_dask_nvt.py ............................................ [  6%]
.....................................................................    [ 15%]
tests/unit/test_io.py .................................................. [ 22%]
..................................................................ssssss [ 31%]
ss..............................................                         [ 38%]
tests/unit/test_notebooks.py ..s..                                       [ 38%]
tests/unit/test_ops.py ................................................. [ 45%]
........................................................................ [ 54%]
........................................................................ [ 63%]
...................................                                      [ 68%]
tests/unit/test_tf_dataloader.py ...................                     [ 71%]
tests/unit/test_tf_layers.py ........................................... [ 76%]
...................................                                      [ 81%]
tests/unit/test_tools.py ......................                          [ 84%]
tests/unit/test_torch_dataloader.py ......F..........FF..........s       [ 88%]
tests/unit/test_workflow.py ............................................ [ 93%]
...............................................                          [100%]

=================================== FAILURES ===================================
_________ test_empty_cols[label_name0-cont_names0-cat_names0-parquet] __________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-41/test_empty_cols_label_name0_co0')
df =       name-cat name-string    id  label         x         y
0     Patricia     Michael   990   1002 -0.005722  0.54568...n  1004   1075  0.821929 -0.615211
4320       Ray      Xavier  1007    986 -0.643196  0.034432

[4321 rows x 6 columns]
dataset = <nvtabular.io.dataset.Dataset object at 0x7f7cbc73d070>
engine = 'parquet', cat_names = ['name-cat', 'name-string']
cont_names = ['x', 'y', 'id'], label_name = ['label']

    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("cat_names", [["name-cat", "name-string"], []])
    @pytest.mark.parametrize("cont_names", [["x", "y", "id"], []])
    @pytest.mark.parametrize("label_name", [["label"], []])
    def test_empty_cols(tmpdir, df, dataset, engine, cat_names, cont_names, label_name):
    
        features = []
        if cont_names:
            features.append(cont_names >> ops.FillMedian() >> ops.Normalize())
        if cat_names:
            features.append(cat_names >> ops.Categorify())
    
        # test out https://github.com/NVIDIA/NVTabular/issues/149 making sure we can iterate over
        # empty cats/conts
        graph = sum(features, nvt.ColumnGroup(label_name))
        if not graph.columns:
            # if we don't have conts/cats/labels we're done
            return
    
        processor = nvt.Workflow(sum(features, nvt.ColumnGroup(label_name)))
    
        output_train = os.path.join(tmpdir, "train/")
        os.mkdir(output_train)
    
>       df_out = processor.fit_transform(dataset).to_ddf().compute(scheduler="synchronous")

tests/unit/test_torch_dataloader.py:73: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:177: in fit_transform
    self.fit(dataset)
nvtabular/workflow.py:147: in fit
    op.fit_finalize(computed_stats)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <nvtabular.ops.fill.FillMedian object at 0x7f7cbc73dbe0>
dask_stats = x             <NA>
y     -0.020037743
id          1000.0
Name: 0.5, dtype: float64

    @annotate("FillMedian_finalize", color="green", domain="nvt_python")
    def fit_finalize(self, dask_stats):
        index = dask_stats.index
        vals = index.values_host if hasattr(index, "values_host") else index.values
        for col in vals:
>           self.medians[col] = float(dask_stats[col])
E           TypeError: float() argument must be a string or a number, not '_NAType'

nvtabular/ops/fill.py:112: TypeError
______________________ test_gpu_dl[None-parquet-10-0.06] _______________________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-41/test_gpu_dl_None_parquet_10_0_1')
df =       name-cat name-string    id  label         x         y
0     Patricia     Michael   990   1002 -0.005722  0.54568...n  1004   1075  0.821929 -0.615211
4320       Ray      Xavier  1007    986 -0.643196  0.034432

[4321 rows x 6 columns]
dataset = <nvtabular.io.dataset.Dataset object at 0x7f7d92066eb0>
batch_size = 10, part_mem_fraction = 0.06, engine = 'parquet', devices = None

    @pytest.mark.parametrize("part_mem_fraction", [0.001, 0.06])
    @pytest.mark.parametrize("batch_size", [1, 10, 100])
    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("devices", [None, GPU_DEVICE_IDS[:2]])
    def test_gpu_dl(tmpdir, df, dataset, batch_size, part_mem_fraction, engine, devices):
        cat_names = ["name-cat", "name-string"]
        cont_names = ["x", "y", "id"]
        label_name = ["label"]
    
        conts = cont_names >> ops.FillMedian() >> ops.Normalize()
        cats = cat_names >> ops.Categorify()
    
        processor = nvt.Workflow(conts + cats + label_name)
    
        output_train = os.path.join(tmpdir, "train/")
        os.mkdir(output_train)
    
>       processor.fit_transform(dataset).to_parquet(
            shuffle=nvt.io.Shuffle.PER_PARTITION,
            output_path=output_train,
            out_files_per_proc=2,
        )

tests/unit/test_torch_dataloader.py:106: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:177: in fit_transform
    self.fit(dataset)
nvtabular/workflow.py:147: in fit
    op.fit_finalize(computed_stats)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <nvtabular.ops.fill.FillMedian object at 0x7f7da2135e50>
dask_stats = x             <NA>
y     -0.020037743
id          1000.0
Name: 0.5, dtype: float64

    @annotate("FillMedian_finalize", color="green", domain="nvt_python")
    def fit_finalize(self, dask_stats):
        index = dask_stats.index
        vals = index.values_host if hasattr(index, "values_host") else index.values
        for col in vals:
>           self.medians[col] = float(dask_stats[col])
E           TypeError: float() argument must be a string or a number, not '_NAType'

nvtabular/ops/fill.py:112: TypeError
_____________________ test_gpu_dl[None-parquet-100-0.001] ______________________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-41/test_gpu_dl_None_parquet_100_00')
df =       name-cat name-string    id  label         x         y
0     Patricia     Michael   990   1002 -0.005722  0.54568...n  1004   1075  0.821929 -0.615211
4320       Ray      Xavier  1007    986 -0.643196  0.034432

[4321 rows x 6 columns]
dataset = <nvtabular.io.dataset.Dataset object at 0x7f7da21238e0>
batch_size = 100, part_mem_fraction = 0.001, engine = 'parquet', devices = None

    @pytest.mark.parametrize("part_mem_fraction", [0.001, 0.06])
    @pytest.mark.parametrize("batch_size", [1, 10, 100])
    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("devices", [None, GPU_DEVICE_IDS[:2]])
    def test_gpu_dl(tmpdir, df, dataset, batch_size, part_mem_fraction, engine, devices):
        cat_names = ["name-cat", "name-string"]
        cont_names = ["x", "y", "id"]
        label_name = ["label"]
    
        conts = cont_names >> ops.FillMedian() >> ops.Normalize()
        cats = cat_names >> ops.Categorify()
    
        processor = nvt.Workflow(conts + cats + label_name)
    
        output_train = os.path.join(tmpdir, "train/")
        os.mkdir(output_train)
    
>       processor.fit_transform(dataset).to_parquet(
            shuffle=nvt.io.Shuffle.PER_PARTITION,
            output_path=output_train,
            out_files_per_proc=2,
        )

tests/unit/test_torch_dataloader.py:106: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:177: in fit_transform
    self.fit(dataset)
nvtabular/workflow.py:147: in fit
    op.fit_finalize(computed_stats)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <nvtabular.ops.fill.FillMedian object at 0x7f7d920667f0>
dask_stats = x             <NA>
y     -0.020037743
id          1000.0
Name: 0.5, dtype: float64

    @annotate("FillMedian_finalize", color="green", domain="nvt_python")
    def fit_finalize(self, dask_stats):
        index = dask_stats.index
        vals = index.values_host if hasattr(index, "values_host") else index.values
        for col in vals:
>           self.medians[col] = float(dask_stats[col])
E           TypeError: float() argument must be a string or a number, not '_NAType'

nvtabular/ops/fill.py:112: TypeError

----------- coverage: platform linux, python 3.8.8-final-0 -----------
Name                                                           Stmts   Miss Branch BrPart  Cover   Missing
----------------------------------------------------------------------------------------------------------
nvtabular/__init__.py                                             12      0      0      0   100%
nvtabular/column_group.py                                        149     18     80      5    86%   54, 87, 128, 151-164, 191, 278
nvtabular/dispatch.py                                             81     11     38      5    83%   35, 45->47, 69, 94, 111, 118, 135-138, 167-170
nvtabular/framework_utils/__init__.py                              0      0      0      0   100%
nvtabular/framework_utils/tensorflow/__init__.py                   1      0      0      0   100%
nvtabular/framework_utils/tensorflow/feature_column_utils.py     146    137     96      0     4%   28-32, 69-303
nvtabular/framework_utils/tensorflow/layers/__init__.py            4      0      0      0   100%
nvtabular/framework_utils/tensorflow/layers/embedding.py         153     14     89      7    87%   56, 64->45, 100, 108, 188, 240-248, 251, 344->352, 366->369, 372-373, 376
nvtabular/framework_utils/tensorflow/layers/interaction.py        47     25     20      1    43%   49, 74-103, 106-110, 113
nvtabular/framework_utils/tensorflow/layers/outer_product.py      30     24     10      0    15%   37-38, 41-60, 71-84, 87
nvtabular/framework_utils/torch/__init__.py                        0      0      0      0   100%
nvtabular/framework_utils/torch/layers/__init__.py                 2      0      0      0   100%
nvtabular/framework_utils/torch/layers/embeddings.py              27     12     12      1    51%   47, 71-79, 82-88
nvtabular/framework_utils/torch/models.py                         41      6     22      7    76%   56, 59, 64, 85, 87-88, 91->94, 98->100
nvtabular/framework_utils/torch/utils.py                          31      7     10      3    76%   52, 56-58, 67-69
nvtabular/inference/__init__.py                                    0      0      0      0   100%
nvtabular/inference/triton/__init__.py                           218    211     90      0     2%   24-494
nvtabular/inference/triton/model.py                               56     56     22      0     0%   27-142
nvtabular/inference/triton/model_hugectr.py                       44     44     14      0     0%   27-120
nvtabular/io/__init__.py                                           4      0      0      0   100%
nvtabular/io/avro.py                                              88     88     30      0     0%   16-189
nvtabular/io/csv.py                                               54      4     20      5    88%   95, 99->103, 104, 106, 120
nvtabular/io/dask.py                                             178      7     68     11    93%   109, 112, 148, 223, 380->378, 408->411, 419, 423->425, 425->421, 430, 432
nvtabular/io/dataframe_engine.py                                  58      2     30      6    91%   44, 82->86, 86->91, 88->91, 91->110, 119
nvtabular/io/dataset.py                                          245     30    112     20    85%   254, 256, 269, 278, 296-310, 409->474, 414-417, 422->432, 427-428, 439->437, 453->457, 503, 604->606, 606->615, 616, 623-624, 630, 636, 731-732, 844-849, 855, 880
nvtabular/io/dataset_engine.py                                    23      1      0      0    96%   45
nvtabular/io/hugectr.py                                           45      2     24      2    91%   34, 74->97, 101
nvtabular/io/parquet.py                                          472     27    150     12    93%   83-91, 136-143, 184-186, 311-316, 354-359, 475->482, 545->550, 551-552, 672, 676, 680, 718, 735, 739, 746->748, 866->871, 876->886, 913
nvtabular/io/shuffle.py                                           30      7     12      2    69%   41-48
nvtabular/io/writer.py                                           169     12     64      5    92%   31, 48, 76, 122, 125, 202, 211, 214, 257, 278-280
nvtabular/io/writer_factory.py                                    18      2      8      2    85%   35, 60
nvtabular/loader/__init__.py                                       0      0      0      0   100%
nvtabular/loader/backend.py                                      269     15    112      9    94%   77-78, 107, 124, 131-132, 143, 203->205, 220, 243-244, 283->287, 358, 362-363, 457, 464
nvtabular/loader/tensorflow.py                                   120      8     52      7    90%   52, 60-63, 73, 83, 282, 309->313, 342
nvtabular/loader/tf_utils.py                                      55     10     20      5    80%   29->32, 32->34, 39->41, 43, 50-51, 58-60, 66-70
nvtabular/loader/torch.py                                         43     10      8      0    69%   25-27, 30-36
nvtabular/ops/__init__.py                                         19      0      0      0   100%
nvtabular/ops/bucketize.py                                        25      4     16      2    76%   46, 49-52
nvtabular/ops/categorify.py                                      513     69    298     47    83%   223, 240, 244, 252, 260, 262, 284, 303-304, 336-337, 399-401, 459-461, 466->468, 541, 579, 608->611, 612-614, 621-622, 635-637, 638->606, 654, 664, 666, 672, 688-689, 694, 697->700, 710, 734-736, 739, 741->743, 755-758, 784, 788, 790, 802-805, 920, 922, 964->985, 970->985, 986-991, 1028, 1044->1049, 1048, 1058->1055, 1063->1055, 1071, 1079-1089
nvtabular/ops/clip.py                                             19      2      6      3    80%   45, 53->55, 56
nvtabular/ops/column_similarity.py                                88     22     32      5    69%   84, 154-155, 164-166, 174-190, 205->215, 207->210, 211, 221
nvtabular/ops/data_stats.py                                       57      1     24      4    94%   88->90, 90->92, 93->84, 96->84, 101
nvtabular/ops/difference_lag.py                                   26      0      8      1    97%   67->69
nvtabular/ops/dropna.py                                            9      0      0      0   100%
nvtabular/ops/fill.py                                             58      2     20      1    96%   93, 119
nvtabular/ops/filter.py                                           21      1      6      1    93%   44
nvtabular/ops/hash_bucket.py                                      32      2     18      2    88%   73, 102
nvtabular/ops/hashed_cross.py                                     29      3     13      4    83%   51, 66, 80->exit, 81
nvtabular/ops/join_external.py                                    69      5     28      6    89%   96, 98, 116, 129->133, 153, 168
nvtabular/ops/join_groupby.py                                     80      5     28      2    94%   106, 109->116, 183-184, 187-188
nvtabular/ops/lambdaop.py                                         27      3     10      3    84%   61, 65, 78
nvtabular/ops/logop.py                                             9      0      0      0   100%
nvtabular/ops/moments.py                                          65      0     20      0   100%
nvtabular/ops/normalize.py                                        64      6     14      2    87%   61->60, 67-68, 100-101, 123-124
nvtabular/ops/operator.py                                         15      1      2      1    88%   24
nvtabular/ops/rename.py                                           18      3     10      3    71%   41, 54, 58
nvtabular/ops/stat_operator.py                                    11      0      0      0   100%
nvtabular/ops/target_encoding.py                                 151     11     66      5    91%   143, 163->167, 170->179, 222-223, 226-227, 236-242, 335->338
nvtabular/tools/__init__.py                                        0      0      0      0   100%
nvtabular/tools/data_gen.py                                      235      1     62      4    98%   183->185, 230->234, 320->319, 322
nvtabular/tools/dataset_inspector.py                              49      9     18      0    75%   29-38
nvtabular/tools/inspector_script.py                               46     46      0      0     0%   17-168
nvtabular/utils.py                                                42     16     18      7    55%   20-21, 25-26, 35, 39, 48-51, 53-55, 58, 61, 67, 73
nvtabular/worker.py                                               68      1     30      2    97%   73, 83->98
nvtabular/workflow.py                                            140      9     62      4    93%   39, 125, 137-139, 242, 270-271, 341
----------------------------------------------------------------------------------------------------------
TOTAL                                                           4898   1012   2022    224    77%
Coverage XML written to file coverage.xml

Required test coverage of 70% reached. Total coverage: 76.62%
=========================== short test summary info ============================
FAILED tests/unit/test_torch_dataloader.py::test_empty_cols[label_name0-cont_names0-cat_names0-parquet]
FAILED tests/unit/test_torch_dataloader.py::test_gpu_dl[None-parquet-10-0.06]
FAILED tests/unit/test_torch_dataloader.py::test_gpu_dl[None-parquet-100-0.001]
============ 3 failed, 750 passed, 12 skipped in 587.86s (0:09:47) =============
Build step 'Execute shell' marked build as failure
Performing Post build task...
Match found for : : True
Logical operation result is TRUE
Running script  : #!/bin/bash
source activate rapids
cd /var/jenkins_home/
CUDA_VISIBLE_DEVICES=1 python test_res_push.py "https://api.GitHub.com/repos/NVIDIA/NVTabular/issues/$ghprbPullId/comments" "/var/jenkins_home/jobs/$JOB_NAME/builds/$BUILD_NUMBER/log" 
[nvtabular_tests] $ /bin/bash /tmp/jenkins7476407945782677737.sh
</pre></details>
<details><summary>Click to view CI Results</summary><pre>GitHub pull request #677 of commit 89f29ce11125543cc1b4ca94b74cbbe0d4583adc, no merge conflicts.
Running as SYSTEM
Setting status of 89f29ce11125543cc1b4ca94b74cbbe0d4583adc to PENDING with url http://10.20.13.93:8080/job/nvtabular_tests/1990/ and message: 'Pending'
Using context: Jenkins Unit Test Run
Building in workspace /var/jenkins_home/workspace/nvtabular_tests
using credential ghub_token
Cloning the remote Git repository
Cloning repository https://github.com/NVIDIA/NVTabular.git
 > git init /var/jenkins_home/workspace/nvtabular_tests/nvtabular # timeout=10
Fetching upstream changes from https://github.com/NVIDIA/NVTabular.git
 > git --version # timeout=10
using GIT_ASKPASS to set credentials github token setup
 > git fetch --tags --force --progress -- https://github.com/NVIDIA/NVTabular.git +refs/heads/*:refs/remotes/origin/* # timeout=10
 > git config remote.origin.url https://github.com/NVIDIA/NVTabular.git # timeout=10
 > git config --add remote.origin.fetch +refs/heads/*:refs/remotes/origin/* # timeout=10
 > git config remote.origin.url https://github.com/NVIDIA/NVTabular.git # timeout=10
Fetching upstream changes from https://github.com/NVIDIA/NVTabular.git
using GIT_ASKPASS to set credentials github token setup
 > git fetch --tags --force --progress -- https://github.com/NVIDIA/NVTabular.git +refs/pull/677/*:refs/remotes/origin/pr/677/* # timeout=10
 > git rev-parse 89f29ce11125543cc1b4ca94b74cbbe0d4583adc^{commit} # timeout=10
Checking out Revision 89f29ce11125543cc1b4ca94b74cbbe0d4583adc (detached)
 > git config core.sparsecheckout # timeout=10
 > git checkout -f 89f29ce11125543cc1b4ca94b74cbbe0d4583adc # timeout=10
Commit message: "Merge branch 'main' into hive-partitioning"
 > git rev-list --no-walk 9fb2dd5d8fdc927f73405b5819e1ad46d6d9a4a3 # timeout=10
[nvtabular_tests] $ /bin/bash /tmp/jenkins1838599612458303638.sh
Obtaining file:///var/jenkins_home/workspace/nvtabular_tests/nvtabular
  Installing build dependencies: started
  Installing build dependencies: finished with status 'done'
  Getting requirements to build wheel: started
  Getting requirements to build wheel: finished with status 'done'
    Preparing wheel metadata: started
    Preparing wheel metadata: finished with status 'done'
Installing collected packages: nvtabular
  Running setup.py develop for nvtabular
Successfully installed nvtabular
All done! ✨ 🍰 ✨
94 files would be left unchanged.
/conda/envs/rapids/lib/python3.8/site-packages/isort/files.py:30: UserWarning: Likely recursive symlink detected to /var/jenkins_home/workspace/nvtabular_tests/nvtabular/images
  warn(f"Likely recursive symlink detected to {resolved_path}")
============================= test session starts ==============================
platform linux -- Python 3.8.8, pytest-6.2.2, py-1.10.0, pluggy-0.13.1
rootdir: /var/jenkins_home/workspace/nvtabular_tests/nvtabular, configfile: pyproject.toml
plugins: cov-2.11.1, xdist-2.2.1, forked-1.3.0
collected 770 items / 2 skipped / 768 selected

tests/unit/test_column_group.py .                                        [  0%]
tests/unit/test_column_similarity.py ......                              [  0%]
tests/unit/test_dask_nvt.py ............................................ [  6%]
.....................................................................    [ 15%]
tests/unit/test_io.py .................................................. [ 22%]
..................................................................ssssss [ 31%]
ss..............................................                         [ 37%]
tests/unit/test_notebooks.py ..s..                                       [ 38%]
tests/unit/test_ops.py ................................................. [ 44%]
........................................................................ [ 54%]
........................................................................ [ 63%]
...................................                                      [ 67%]
tests/unit/test_tf_dataloader.py .........................s              [ 71%]
tests/unit/test_tf_layers.py ........................................... [ 76%]
...................................                                      [ 81%]
tests/unit/test_tools.py ......................                          [ 84%]
tests/unit/test_torch_dataloader.py .............................s       [ 88%]
tests/unit/test_workflow.py ............................................ [ 93%]
...............................................                          [100%]

----------- coverage: platform linux, python 3.8.8-final-0 -----------
Name                                                           Stmts   Miss Branch BrPart  Cover   Missing
----------------------------------------------------------------------------------------------------------
nvtabular/__init__.py                                             12      0      0      0   100%
nvtabular/column_group.py                                        149     18     80      5    86%   54, 87, 128, 151-164, 191, 278
nvtabular/dispatch.py                                             81     11     38      5    83%   35, 45->47, 69, 94, 111, 118, 135-138, 167-170
nvtabular/framework_utils/__init__.py                              0      0      0      0   100%
nvtabular/framework_utils/tensorflow/__init__.py                   1      0      0      0   100%
nvtabular/framework_utils/tensorflow/feature_column_utils.py     146    137     96      0     4%   28-32, 69-303
nvtabular/framework_utils/tensorflow/layers/__init__.py            4      0      0      0   100%
nvtabular/framework_utils/tensorflow/layers/embedding.py         153     14     89      7    87%   56, 64->45, 100, 108, 188, 240-248, 251, 344->352, 366->369, 372-373, 376
nvtabular/framework_utils/tensorflow/layers/interaction.py        47     25     20      1    43%   49, 74-103, 106-110, 113
nvtabular/framework_utils/tensorflow/layers/outer_product.py      30     24     10      0    15%   37-38, 41-60, 71-84, 87
nvtabular/framework_utils/torch/__init__.py                        0      0      0      0   100%
nvtabular/framework_utils/torch/layers/__init__.py                 2      0      0      0   100%
nvtabular/framework_utils/torch/layers/embeddings.py              27     12     12      1    51%   47, 71-79, 82-88
nvtabular/framework_utils/torch/models.py                         41      6     22      7    76%   56, 59, 64, 85, 87-88, 91->94, 98->100
nvtabular/framework_utils/torch/utils.py                          31      7     10      3    76%   52, 56-58, 67-69
nvtabular/inference/__init__.py                                    0      0      0      0   100%
nvtabular/inference/triton/__init__.py                           218    211     90      0     2%   24-494
nvtabular/inference/triton/model.py                               56     56     22      0     0%   27-142
nvtabular/inference/triton/model_hugectr.py                       44     44     14      0     0%   27-120
nvtabular/io/__init__.py                                           4      0      0      0   100%
nvtabular/io/avro.py                                              88     88     30      0     0%   16-189
nvtabular/io/csv.py                                               54      4     20      5    88%   95, 99->103, 104, 106, 120
nvtabular/io/dask.py                                             178      7     68     11    93%   109, 112, 148, 223, 380->378, 408->411, 419, 423->425, 425->421, 430, 432
nvtabular/io/dataframe_engine.py                                  58      3     30      6    90%   44, 63, 82->86, 86->91, 88->91, 91->110, 119
nvtabular/io/dataset.py                                          245     30    112     20    85%   254, 256, 269, 278, 296-310, 409->474, 414-417, 422->432, 427-428, 439->437, 453->457, 503, 604->606, 606->615, 616, 623-624, 630, 636, 731-732, 844-849, 855, 880
nvtabular/io/dataset_engine.py                                    23      1      0      0    96%   45
nvtabular/io/hugectr.py                                           45      2     24      2    91%   34, 74->97, 101
nvtabular/io/parquet.py                                          472     28    150     13    93%   83-91, 136-143, 155, 184-186, 311-316, 354-359, 475->482, 545->550, 551-552, 672, 676, 680, 718, 735, 739, 746->748, 866->871, 876->886, 913
nvtabular/io/shuffle.py                                           30      7     12      2    69%   41-48
nvtabular/io/writer.py                                           169     12     64      5    92%   31, 48, 76, 122, 125, 202, 211, 214, 257, 278-280
nvtabular/io/writer_factory.py                                    18      2      8      2    85%   35, 60
nvtabular/loader/__init__.py                                       0      0      0      0   100%
nvtabular/loader/backend.py                                      278     17    114     10    93%   84-85, 113, 130, 137-138, 149, 214->216, 224-228, 238, 261-262, 301->305, 376, 380-381, 475, 482
nvtabular/loader/tensorflow.py                                   120     11     52      7    88%   52, 60-63, 73, 83, 287, 302-304, 314->318, 347
nvtabular/loader/tf_utils.py                                      55     10     20      5    80%   29->32, 32->34, 39->41, 43, 50-51, 58-60, 66-70
nvtabular/loader/torch.py                                         43     10      8      0    69%   25-27, 30-36
nvtabular/ops/__init__.py                                         19      0      0      0   100%
nvtabular/ops/bucketize.py                                        25      4     16      2    76%   46, 49-52
nvtabular/ops/categorify.py                                      513     69    298     47    83%   223, 240, 244, 252, 260, 262, 284, 303-304, 336-337, 399-401, 459-461, 466->468, 541, 579, 608->611, 612-614, 621-622, 635-637, 638->606, 654, 664, 666, 672, 688-689, 694, 697->700, 710, 734-736, 739, 741->743, 755-758, 784, 788, 790, 802-805, 920, 922, 964->985, 970->985, 986-991, 1028, 1044->1049, 1048, 1058->1055, 1063->1055, 1071, 1079-1089
nvtabular/ops/clip.py                                             19      2      6      3    80%   45, 53->55, 56
nvtabular/ops/column_similarity.py                                88     22     32      5    69%   84, 154-155, 164-166, 174-190, 205->215, 207->210, 211, 221
nvtabular/ops/data_stats.py                                       57      1     24      4    94%   88->90, 90->92, 93->84, 96->84, 101
nvtabular/ops/difference_lag.py                                   26      0      8      1    97%   67->69
nvtabular/ops/dropna.py                                            9      0      0      0   100%
nvtabular/ops/fill.py                                             58      2     20      1    96%   93, 119
nvtabular/ops/filter.py                                           21      1      6      1    93%   44
nvtabular/ops/hash_bucket.py                                      32      2     18      2    88%   73, 102
nvtabular/ops/hashed_cross.py                                     29      3     13      4    83%   51, 66, 80->exit, 81
nvtabular/ops/join_external.py                                    69      5     28      6    89%   96, 98, 116, 129->133, 153, 168
nvtabular/ops/join_groupby.py                                     80      5     28      2    94%   106, 109->116, 183-184, 187-188
nvtabular/ops/lambdaop.py                                         27      3     10      3    84%   61, 65, 78
nvtabular/ops/logop.py                                             9      0      0      0   100%
nvtabular/ops/moments.py                                          65      0     20      0   100%
nvtabular/ops/normalize.py                                        64      6     14      2    87%   61->60, 67-68, 100-101, 123-124
nvtabular/ops/operator.py                                         15      1      2      1    88%   24
nvtabular/ops/rename.py                                           18      3     10      3    71%   41, 54, 58
nvtabular/ops/stat_operator.py                                    11      0      0      0   100%
nvtabular/ops/target_encoding.py                                 151     11     66      5    91%   143, 163->167, 170->179, 222-223, 226-227, 236-242, 335->338
nvtabular/tools/__init__.py                                        0      0      0      0   100%
nvtabular/tools/data_gen.py                                      235      1     62      4    98%   183->185, 230->234, 320->319, 322
nvtabular/tools/dataset_inspector.py                              49      9     18      0    75%   29-38
nvtabular/tools/inspector_script.py                               46     46      0      0     0%   17-168
nvtabular/utils.py                                                42     16     18      7    55%   20-21, 25-26, 35, 39, 48-51, 53-55, 58, 61, 67, 73
nvtabular/worker.py                                               68      1     30      2    97%   73, 83->98
nvtabular/workflow.py                                            140      9     62      4    93%   39, 125, 137-139, 242, 270-271, 341
----------------------------------------------------------------------------------------------------------
TOTAL                                                           4907   1019   2024    226    77%
Coverage XML written to file coverage.xml

Required test coverage of 70% reached. Total coverage: 76.53%

================= 759 passed, 13 skipped in 596.36s (0:09:56) ==================
Performing Post build task...
Match found for : : True
Logical operation result is TRUE
Running script  : #!/bin/bash
source activate rapids
cd /var/jenkins_home/
CUDA_VISIBLE_DEVICES=1 python test_res_push.py "https://api.GitHub.com/repos/NVIDIA/NVTabular/issues/$ghprbPullId/comments" "/var/jenkins_home/jobs/$JOB_NAME/builds/$BUILD_NUMBER/log" 
[nvtabular_tests] $ /bin/bash /tmp/jenkins5509550321025932911.sh
</pre></details>
Sounds great @rjzamora ! This PR will allow incremental training and evaluation of sequential recommender models and time series, as allows splitting data by time windows.
<details><summary>Click to view CI Results</summary><pre>GitHub pull request #677 of commit 0eedde3f05679f9948ceda0136fbe06b704e6c6b, no merge conflicts.
Running as SYSTEM
Setting status of 0eedde3f05679f9948ceda0136fbe06b704e6c6b to PENDING with url http://10.20.13.93:8080/job/nvtabular_tests/1999/ and message: 'Pending'
Using context: Jenkins Unit Test Run
Building in workspace /var/jenkins_home/workspace/nvtabular_tests
using credential ghub_token
Cloning the remote Git repository
Cloning repository https://github.com/NVIDIA/NVTabular.git
 > git init /var/jenkins_home/workspace/nvtabular_tests/nvtabular # timeout=10
Fetching upstream changes from https://github.com/NVIDIA/NVTabular.git
 > git --version # timeout=10
using GIT_ASKPASS to set credentials github token setup
 > git fetch --tags --force --progress -- https://github.com/NVIDIA/NVTabular.git +refs/heads/*:refs/remotes/origin/* # timeout=10
 > git config remote.origin.url https://github.com/NVIDIA/NVTabular.git # timeout=10
 > git config --add remote.origin.fetch +refs/heads/*:refs/remotes/origin/* # timeout=10
 > git config remote.origin.url https://github.com/NVIDIA/NVTabular.git # timeout=10
Fetching upstream changes from https://github.com/NVIDIA/NVTabular.git
using GIT_ASKPASS to set credentials github token setup
 > git fetch --tags --force --progress -- https://github.com/NVIDIA/NVTabular.git +refs/pull/677/*:refs/remotes/origin/pr/677/* # timeout=10
 > git rev-parse 0eedde3f05679f9948ceda0136fbe06b704e6c6b^{commit} # timeout=10
Checking out Revision 0eedde3f05679f9948ceda0136fbe06b704e6c6b (detached)
 > git config core.sparsecheckout # timeout=10
 > git checkout -f 0eedde3f05679f9948ceda0136fbe06b704e6c6b # timeout=10
Commit message: "Merge branch 'main' into hive-partitioning"
 > git rev-list --no-walk 2dcc4a9ed07a2ff8254449e1291ebc2e4281ddbf # timeout=10
First time build. Skipping changelog.
[nvtabular_tests] $ /bin/bash /tmp/jenkins5306605500818761616.sh
Obtaining file:///var/jenkins_home/workspace/nvtabular_tests/nvtabular
  Installing build dependencies: started
  Installing build dependencies: finished with status 'done'
  Getting requirements to build wheel: started
  Getting requirements to build wheel: finished with status 'done'
    Preparing wheel metadata: started
    Preparing wheel metadata: finished with status 'done'
Installing collected packages: nvtabular
  Running setup.py develop for nvtabular
Successfully installed nvtabular
All done! ✨ 🍰 ✨
94 files would be left unchanged.
/conda/envs/rapids/lib/python3.8/site-packages/isort/files.py:30: UserWarning: Likely recursive symlink detected to /var/jenkins_home/workspace/nvtabular_tests/nvtabular/images
  warn(f"Likely recursive symlink detected to {resolved_path}")
============================= test session starts ==============================
platform linux -- Python 3.8.8, pytest-6.2.2, py-1.10.0, pluggy-0.13.1
rootdir: /var/jenkins_home/workspace/nvtabular_tests/nvtabular, configfile: pyproject.toml
plugins: cov-2.11.1, xdist-2.2.1, forked-1.3.0
collected 770 items / 2 skipped / 768 selected

tests/unit/test_column_group.py .                                        [  0%]
tests/unit/test_column_similarity.py ......                              [  0%]
tests/unit/test_dask_nvt.py ............................................ [  6%]
.....................................................................    [ 15%]
tests/unit/test_io.py .................................................. [ 22%]
..................................................................ssssss [ 31%]
ss..............................................                         [ 37%]
tests/unit/test_notebooks.py ..s..                                       [ 38%]
tests/unit/test_ops.py ................................................. [ 44%]
........................................................................ [ 54%]
........................................................................ [ 63%]
...................................                                      [ 67%]
tests/unit/test_tf_dataloader.py ..FFFFFFFFFFF............s              [ 71%]
tests/unit/test_tf_layers.py ........................................... [ 76%]
...................................                                      [ 81%]
tests/unit/test_tools.py ......................                          [ 84%]
tests/unit/test_torch_dataloader.py .............................s       [ 88%]
tests/unit/test_workflow.py ............................................ [ 93%]
...............................................                          [100%]

=================================== FAILURES ===================================
_____________________ test_tf_gpu_dl[True-1-parquet-0.06] ______________________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-49/test_tf_gpu_dl_True_1_parquet_1')
paths = ['/tmp/pytest-of-jenkins/pytest-49/parquet0/dataset-0.parquet', '/tmp/pytest-of-jenkins/pytest-49/parquet0/dataset-1.parquet']
use_paths = True
dataset = <nvtabular.io.dataset.Dataset object at 0x7febf0c63e50>
batch_size = 1, gpu_memory_frac = 0.06, engine = 'parquet'

    @pytest.mark.parametrize("gpu_memory_frac", [0.01, 0.06])
    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("batch_size", [1, 10, 100])
    @pytest.mark.parametrize("use_paths", [True, False])
    def test_tf_gpu_dl(tmpdir, paths, use_paths, dataset, batch_size, gpu_memory_frac, engine):
        cont_names = ["x", "y", "id"]
        cat_names = ["name-string"]
        label_name = ["label"]
        if engine == "parquet":
            cat_names.append("name-cat")
    
        columns = cont_names + cat_names
    
        conts = cont_names >> ops.FillMedian() >> ops.Normalize()
        cats = cat_names >> ops.Categorify()
    
        workflow = nvt.Workflow(conts + cats + label_name)
>       workflow.fit(dataset)

tests/unit/test_tf_dataloader.py:91: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:147: in fit
    op.fit_finalize(computed_stats)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <nvtabular.ops.fill.FillMedian object at 0x7febf0c63310>
dask_stats = x       <NA>
y       <NA>
id    1001.0
Name: 0.5, dtype: float64

    @annotate("FillMedian_finalize", color="green", domain="nvt_python")
    def fit_finalize(self, dask_stats):
        index = dask_stats.index
        vals = index.values_host if hasattr(index, "values_host") else index.values
        for col in vals:
>           self.medians[col] = float(dask_stats[col])
E           TypeError: float() argument must be a string or a number, not '_NAType'

nvtabular/ops/fill.py:112: TypeError
_____________________ test_tf_gpu_dl[True-10-parquet-0.01] _____________________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-49/test_tf_gpu_dl_True_10_parquet0')
paths = ['/tmp/pytest-of-jenkins/pytest-49/parquet0/dataset-0.parquet', '/tmp/pytest-of-jenkins/pytest-49/parquet0/dataset-1.parquet']
use_paths = True
dataset = <nvtabular.io.dataset.Dataset object at 0x7fec887571f0>
batch_size = 10, gpu_memory_frac = 0.01, engine = 'parquet'

    @pytest.mark.parametrize("gpu_memory_frac", [0.01, 0.06])
    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("batch_size", [1, 10, 100])
    @pytest.mark.parametrize("use_paths", [True, False])
    def test_tf_gpu_dl(tmpdir, paths, use_paths, dataset, batch_size, gpu_memory_frac, engine):
        cont_names = ["x", "y", "id"]
        cat_names = ["name-string"]
        label_name = ["label"]
        if engine == "parquet":
            cat_names.append("name-cat")
    
        columns = cont_names + cat_names
    
        conts = cont_names >> ops.FillMedian() >> ops.Normalize()
        cats = cat_names >> ops.Categorify()
    
        workflow = nvt.Workflow(conts + cats + label_name)
>       workflow.fit(dataset)

tests/unit/test_tf_dataloader.py:91: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:147: in fit
    op.fit_finalize(computed_stats)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <nvtabular.ops.fill.FillMedian object at 0x7febf0c630a0>
dask_stats = x       <NA>
y       <NA>
id    1001.0
Name: 0.5, dtype: float64

    @annotate("FillMedian_finalize", color="green", domain="nvt_python")
    def fit_finalize(self, dask_stats):
        index = dask_stats.index
        vals = index.values_host if hasattr(index, "values_host") else index.values
        for col in vals:
>           self.medians[col] = float(dask_stats[col])
E           TypeError: float() argument must be a string or a number, not '_NAType'

nvtabular/ops/fill.py:112: TypeError
_____________________ test_tf_gpu_dl[True-10-parquet-0.06] _____________________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-49/test_tf_gpu_dl_True_10_parquet1')
paths = ['/tmp/pytest-of-jenkins/pytest-49/parquet0/dataset-0.parquet', '/tmp/pytest-of-jenkins/pytest-49/parquet0/dataset-1.parquet']
use_paths = True
dataset = <nvtabular.io.dataset.Dataset object at 0x7fecbafaa310>
batch_size = 10, gpu_memory_frac = 0.06, engine = 'parquet'

    @pytest.mark.parametrize("gpu_memory_frac", [0.01, 0.06])
    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("batch_size", [1, 10, 100])
    @pytest.mark.parametrize("use_paths", [True, False])
    def test_tf_gpu_dl(tmpdir, paths, use_paths, dataset, batch_size, gpu_memory_frac, engine):
        cont_names = ["x", "y", "id"]
        cat_names = ["name-string"]
        label_name = ["label"]
        if engine == "parquet":
            cat_names.append("name-cat")
    
        columns = cont_names + cat_names
    
        conts = cont_names >> ops.FillMedian() >> ops.Normalize()
        cats = cat_names >> ops.Categorify()
    
        workflow = nvt.Workflow(conts + cats + label_name)
>       workflow.fit(dataset)

tests/unit/test_tf_dataloader.py:91: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:147: in fit
    op.fit_finalize(computed_stats)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <nvtabular.ops.fill.FillMedian object at 0x7fecbafaa7c0>
dask_stats = x     0.017360521
y            <NA>
id         1001.0
Name: 0.5, dtype: float64

    @annotate("FillMedian_finalize", color="green", domain="nvt_python")
    def fit_finalize(self, dask_stats):
        index = dask_stats.index
        vals = index.values_host if hasattr(index, "values_host") else index.values
        for col in vals:
>           self.medians[col] = float(dask_stats[col])
E           TypeError: float() argument must be a string or a number, not '_NAType'

nvtabular/ops/fill.py:112: TypeError
____________________ test_tf_gpu_dl[True-100-parquet-0.01] _____________________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-49/test_tf_gpu_dl_True_100_parque0')
paths = ['/tmp/pytest-of-jenkins/pytest-49/parquet0/dataset-0.parquet', '/tmp/pytest-of-jenkins/pytest-49/parquet0/dataset-1.parquet']
use_paths = True
dataset = <nvtabular.io.dataset.Dataset object at 0x7fecbb15c2e0>
batch_size = 100, gpu_memory_frac = 0.01, engine = 'parquet'

    @pytest.mark.parametrize("gpu_memory_frac", [0.01, 0.06])
    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("batch_size", [1, 10, 100])
    @pytest.mark.parametrize("use_paths", [True, False])
    def test_tf_gpu_dl(tmpdir, paths, use_paths, dataset, batch_size, gpu_memory_frac, engine):
        cont_names = ["x", "y", "id"]
        cat_names = ["name-string"]
        label_name = ["label"]
        if engine == "parquet":
            cat_names.append("name-cat")
    
        columns = cont_names + cat_names
    
        conts = cont_names >> ops.FillMedian() >> ops.Normalize()
        cats = cat_names >> ops.Categorify()
    
        workflow = nvt.Workflow(conts + cats + label_name)
>       workflow.fit(dataset)

tests/unit/test_tf_dataloader.py:91: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:147: in fit
    op.fit_finalize(computed_stats)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <nvtabular.ops.fill.FillMedian object at 0x7fecbb0f15e0>
dask_stats = x           <NA>
y     0.00956418
id        1001.0
Name: 0.5, dtype: float64

    @annotate("FillMedian_finalize", color="green", domain="nvt_python")
    def fit_finalize(self, dask_stats):
        index = dask_stats.index
        vals = index.values_host if hasattr(index, "values_host") else index.values
        for col in vals:
>           self.medians[col] = float(dask_stats[col])
E           TypeError: float() argument must be a string or a number, not '_NAType'

nvtabular/ops/fill.py:112: TypeError
____________________ test_tf_gpu_dl[True-100-parquet-0.06] _____________________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-49/test_tf_gpu_dl_True_100_parque1')
paths = ['/tmp/pytest-of-jenkins/pytest-49/parquet0/dataset-0.parquet', '/tmp/pytest-of-jenkins/pytest-49/parquet0/dataset-1.parquet']
use_paths = True
dataset = <nvtabular.io.dataset.Dataset object at 0x7fecbb067370>
batch_size = 100, gpu_memory_frac = 0.06, engine = 'parquet'

    @pytest.mark.parametrize("gpu_memory_frac", [0.01, 0.06])
    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("batch_size", [1, 10, 100])
    @pytest.mark.parametrize("use_paths", [True, False])
    def test_tf_gpu_dl(tmpdir, paths, use_paths, dataset, batch_size, gpu_memory_frac, engine):
        cont_names = ["x", "y", "id"]
        cat_names = ["name-string"]
        label_name = ["label"]
        if engine == "parquet":
            cat_names.append("name-cat")
    
        columns = cont_names + cat_names
    
        conts = cont_names >> ops.FillMedian() >> ops.Normalize()
        cats = cat_names >> ops.Categorify()
    
        workflow = nvt.Workflow(conts + cats + label_name)
>       workflow.fit(dataset)

tests/unit/test_tf_dataloader.py:91: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:147: in fit
    op.fit_finalize(computed_stats)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <nvtabular.ops.fill.FillMedian object at 0x7fecbb067c10>
dask_stats = x       <NA>
y       <NA>
id    1001.0
Name: 0.5, dtype: float64

    @annotate("FillMedian_finalize", color="green", domain="nvt_python")
    def fit_finalize(self, dask_stats):
        index = dask_stats.index
        vals = index.values_host if hasattr(index, "values_host") else index.values
        for col in vals:
>           self.medians[col] = float(dask_stats[col])
E           TypeError: float() argument must be a string or a number, not '_NAType'

nvtabular/ops/fill.py:112: TypeError
_____________________ test_tf_gpu_dl[False-1-parquet-0.01] _____________________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-49/test_tf_gpu_dl_False_1_parquet0')
paths = ['/tmp/pytest-of-jenkins/pytest-49/parquet0/dataset-0.parquet', '/tmp/pytest-of-jenkins/pytest-49/parquet0/dataset-1.parquet']
use_paths = False
dataset = <nvtabular.io.dataset.Dataset object at 0x7fecbafb3a90>
batch_size = 1, gpu_memory_frac = 0.01, engine = 'parquet'

    @pytest.mark.parametrize("gpu_memory_frac", [0.01, 0.06])
    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("batch_size", [1, 10, 100])
    @pytest.mark.parametrize("use_paths", [True, False])
    def test_tf_gpu_dl(tmpdir, paths, use_paths, dataset, batch_size, gpu_memory_frac, engine):
        cont_names = ["x", "y", "id"]
        cat_names = ["name-string"]
        label_name = ["label"]
        if engine == "parquet":
            cat_names.append("name-cat")
    
        columns = cont_names + cat_names
    
        conts = cont_names >> ops.FillMedian() >> ops.Normalize()
        cats = cat_names >> ops.Categorify()
    
        workflow = nvt.Workflow(conts + cats + label_name)
>       workflow.fit(dataset)

tests/unit/test_tf_dataloader.py:91: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:147: in fit
    op.fit_finalize(computed_stats)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <nvtabular.ops.fill.FillMedian object at 0x7fecbb04f700>
dask_stats = x       <NA>
y       <NA>
id    1001.0
Name: 0.5, dtype: float64

    @annotate("FillMedian_finalize", color="green", domain="nvt_python")
    def fit_finalize(self, dask_stats):
        index = dask_stats.index
        vals = index.values_host if hasattr(index, "values_host") else index.values
        for col in vals:
>           self.medians[col] = float(dask_stats[col])
E           TypeError: float() argument must be a string or a number, not '_NAType'

nvtabular/ops/fill.py:112: TypeError
_____________________ test_tf_gpu_dl[False-1-parquet-0.06] _____________________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-49/test_tf_gpu_dl_False_1_parquet1')
paths = ['/tmp/pytest-of-jenkins/pytest-49/parquet0/dataset-0.parquet', '/tmp/pytest-of-jenkins/pytest-49/parquet0/dataset-1.parquet']
use_paths = False
dataset = <nvtabular.io.dataset.Dataset object at 0x7fec886ae6d0>
batch_size = 1, gpu_memory_frac = 0.06, engine = 'parquet'

    @pytest.mark.parametrize("gpu_memory_frac", [0.01, 0.06])
    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("batch_size", [1, 10, 100])
    @pytest.mark.parametrize("use_paths", [True, False])
    def test_tf_gpu_dl(tmpdir, paths, use_paths, dataset, batch_size, gpu_memory_frac, engine):
        cont_names = ["x", "y", "id"]
        cat_names = ["name-string"]
        label_name = ["label"]
        if engine == "parquet":
            cat_names.append("name-cat")
    
        columns = cont_names + cat_names
    
        conts = cont_names >> ops.FillMedian() >> ops.Normalize()
        cats = cat_names >> ops.Categorify()
    
        workflow = nvt.Workflow(conts + cats + label_name)
>       workflow.fit(dataset)

tests/unit/test_tf_dataloader.py:91: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:147: in fit
    op.fit_finalize(computed_stats)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <nvtabular.ops.fill.FillMedian object at 0x7fec886aec40>
dask_stats = x       <NA>
y       <NA>
id    1001.0
Name: 0.5, dtype: float64

    @annotate("FillMedian_finalize", color="green", domain="nvt_python")
    def fit_finalize(self, dask_stats):
        index = dask_stats.index
        vals = index.values_host if hasattr(index, "values_host") else index.values
        for col in vals:
>           self.medians[col] = float(dask_stats[col])
E           TypeError: float() argument must be a string or a number, not '_NAType'

nvtabular/ops/fill.py:112: TypeError
____________________ test_tf_gpu_dl[False-10-parquet-0.01] _____________________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-49/test_tf_gpu_dl_False_10_parque0')
paths = ['/tmp/pytest-of-jenkins/pytest-49/parquet0/dataset-0.parquet', '/tmp/pytest-of-jenkins/pytest-49/parquet0/dataset-1.parquet']
use_paths = False
dataset = <nvtabular.io.dataset.Dataset object at 0x7fec88757a90>
batch_size = 10, gpu_memory_frac = 0.01, engine = 'parquet'

    @pytest.mark.parametrize("gpu_memory_frac", [0.01, 0.06])
    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("batch_size", [1, 10, 100])
    @pytest.mark.parametrize("use_paths", [True, False])
    def test_tf_gpu_dl(tmpdir, paths, use_paths, dataset, batch_size, gpu_memory_frac, engine):
        cont_names = ["x", "y", "id"]
        cat_names = ["name-string"]
        label_name = ["label"]
        if engine == "parquet":
            cat_names.append("name-cat")
    
        columns = cont_names + cat_names
    
        conts = cont_names >> ops.FillMedian() >> ops.Normalize()
        cats = cat_names >> ops.Categorify()
    
        workflow = nvt.Workflow(conts + cats + label_name)
>       workflow.fit(dataset)

tests/unit/test_tf_dataloader.py:91: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:147: in fit
    op.fit_finalize(computed_stats)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <nvtabular.ops.fill.FillMedian object at 0x7fecbaee6430>
dask_stats = x       <NA>
y       <NA>
id    1001.0
Name: 0.5, dtype: float64

    @annotate("FillMedian_finalize", color="green", domain="nvt_python")
    def fit_finalize(self, dask_stats):
        index = dask_stats.index
        vals = index.values_host if hasattr(index, "values_host") else index.values
        for col in vals:
>           self.medians[col] = float(dask_stats[col])
E           TypeError: float() argument must be a string or a number, not '_NAType'

nvtabular/ops/fill.py:112: TypeError
____________________ test_tf_gpu_dl[False-10-parquet-0.06] _____________________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-49/test_tf_gpu_dl_False_10_parque1')
paths = ['/tmp/pytest-of-jenkins/pytest-49/parquet0/dataset-0.parquet', '/tmp/pytest-of-jenkins/pytest-49/parquet0/dataset-1.parquet']
use_paths = False
dataset = <nvtabular.io.dataset.Dataset object at 0x7febf0dd63d0>
batch_size = 10, gpu_memory_frac = 0.06, engine = 'parquet'

    @pytest.mark.parametrize("gpu_memory_frac", [0.01, 0.06])
    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("batch_size", [1, 10, 100])
    @pytest.mark.parametrize("use_paths", [True, False])
    def test_tf_gpu_dl(tmpdir, paths, use_paths, dataset, batch_size, gpu_memory_frac, engine):
        cont_names = ["x", "y", "id"]
        cat_names = ["name-string"]
        label_name = ["label"]
        if engine == "parquet":
            cat_names.append("name-cat")
    
        columns = cont_names + cat_names
    
        conts = cont_names >> ops.FillMedian() >> ops.Normalize()
        cats = cat_names >> ops.Categorify()
    
        workflow = nvt.Workflow(conts + cats + label_name)
>       workflow.fit(dataset)

tests/unit/test_tf_dataloader.py:91: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:147: in fit
    op.fit_finalize(computed_stats)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <nvtabular.ops.fill.FillMedian object at 0x7febf0d823d0>
dask_stats = x       <NA>
y       <NA>
id    1001.0
Name: 0.5, dtype: float64

    @annotate("FillMedian_finalize", color="green", domain="nvt_python")
    def fit_finalize(self, dask_stats):
        index = dask_stats.index
        vals = index.values_host if hasattr(index, "values_host") else index.values
        for col in vals:
>           self.medians[col] = float(dask_stats[col])
E           TypeError: float() argument must be a string or a number, not '_NAType'

nvtabular/ops/fill.py:112: TypeError
____________________ test_tf_gpu_dl[False-100-parquet-0.01] ____________________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-49/test_tf_gpu_dl_False_100_parqu0')
paths = ['/tmp/pytest-of-jenkins/pytest-49/parquet0/dataset-0.parquet', '/tmp/pytest-of-jenkins/pytest-49/parquet0/dataset-1.parquet']
use_paths = False
dataset = <nvtabular.io.dataset.Dataset object at 0x7febf0d714c0>
batch_size = 100, gpu_memory_frac = 0.01, engine = 'parquet'

    @pytest.mark.parametrize("gpu_memory_frac", [0.01, 0.06])
    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("batch_size", [1, 10, 100])
    @pytest.mark.parametrize("use_paths", [True, False])
    def test_tf_gpu_dl(tmpdir, paths, use_paths, dataset, batch_size, gpu_memory_frac, engine):
        cont_names = ["x", "y", "id"]
        cat_names = ["name-string"]
        label_name = ["label"]
        if engine == "parquet":
            cat_names.append("name-cat")
    
        columns = cont_names + cat_names
    
        conts = cont_names >> ops.FillMedian() >> ops.Normalize()
        cats = cat_names >> ops.Categorify()
    
        workflow = nvt.Workflow(conts + cats + label_name)
>       workflow.fit(dataset)

tests/unit/test_tf_dataloader.py:91: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:147: in fit
    op.fit_finalize(computed_stats)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <nvtabular.ops.fill.FillMedian object at 0x7fecb90ed910>
dask_stats = x           <NA>
y     0.00956418
id        1001.0
Name: 0.5, dtype: float64

    @annotate("FillMedian_finalize", color="green", domain="nvt_python")
    def fit_finalize(self, dask_stats):
        index = dask_stats.index
        vals = index.values_host if hasattr(index, "values_host") else index.values
        for col in vals:
>           self.medians[col] = float(dask_stats[col])
E           TypeError: float() argument must be a string or a number, not '_NAType'

nvtabular/ops/fill.py:112: TypeError
____________________ test_tf_gpu_dl[False-100-parquet-0.06] ____________________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-49/test_tf_gpu_dl_False_100_parqu1')
paths = ['/tmp/pytest-of-jenkins/pytest-49/parquet0/dataset-0.parquet', '/tmp/pytest-of-jenkins/pytest-49/parquet0/dataset-1.parquet']
use_paths = False
dataset = <nvtabular.io.dataset.Dataset object at 0x7fec002e6a90>
batch_size = 100, gpu_memory_frac = 0.06, engine = 'parquet'

    @pytest.mark.parametrize("gpu_memory_frac", [0.01, 0.06])
    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("batch_size", [1, 10, 100])
    @pytest.mark.parametrize("use_paths", [True, False])
    def test_tf_gpu_dl(tmpdir, paths, use_paths, dataset, batch_size, gpu_memory_frac, engine):
        cont_names = ["x", "y", "id"]
        cat_names = ["name-string"]
        label_name = ["label"]
        if engine == "parquet":
            cat_names.append("name-cat")
    
        columns = cont_names + cat_names
    
        conts = cont_names >> ops.FillMedian() >> ops.Normalize()
        cats = cat_names >> ops.Categorify()
    
        workflow = nvt.Workflow(conts + cats + label_name)
>       workflow.fit(dataset)

tests/unit/test_tf_dataloader.py:91: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:147: in fit
    op.fit_finalize(computed_stats)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <nvtabular.ops.fill.FillMedian object at 0x7fec002e63a0>
dask_stats = x     0.017360521
y            <NA>
id         1001.0
Name: 0.5, dtype: float64

    @annotate("FillMedian_finalize", color="green", domain="nvt_python")
    def fit_finalize(self, dask_stats):
        index = dask_stats.index
        vals = index.values_host if hasattr(index, "values_host") else index.values
        for col in vals:
>           self.medians[col] = float(dask_stats[col])
E           TypeError: float() argument must be a string or a number, not '_NAType'

nvtabular/ops/fill.py:112: TypeError

----------- coverage: platform linux, python 3.8.8-final-0 -----------
Name                                                           Stmts   Miss Branch BrPart  Cover   Missing
----------------------------------------------------------------------------------------------------------
nvtabular/__init__.py                                             12      0      0      0   100%
nvtabular/column_group.py                                        149     18     80      5    86%   54, 87, 128, 151-164, 191, 278
nvtabular/dispatch.py                                             81     11     38      5    83%   35, 45->47, 69, 94, 111, 118, 135-138, 167-170
nvtabular/framework_utils/__init__.py                              0      0      0      0   100%
nvtabular/framework_utils/tensorflow/__init__.py                   1      0      0      0   100%
nvtabular/framework_utils/tensorflow/feature_column_utils.py     146    137     96      0     4%   28-32, 69-303
nvtabular/framework_utils/tensorflow/layers/__init__.py            4      0      0      0   100%
nvtabular/framework_utils/tensorflow/layers/embedding.py         153     14     89      7    87%   56, 64->45, 100, 108, 188, 240-248, 251, 344->352, 366->369, 372-373, 376
nvtabular/framework_utils/tensorflow/layers/interaction.py        47     25     20      1    43%   49, 74-103, 106-110, 113
nvtabular/framework_utils/tensorflow/layers/outer_product.py      30     24     10      0    15%   37-38, 41-60, 71-84, 87
nvtabular/framework_utils/torch/__init__.py                        0      0      0      0   100%
nvtabular/framework_utils/torch/layers/__init__.py                 2      0      0      0   100%
nvtabular/framework_utils/torch/layers/embeddings.py              27     12     12      1    51%   47, 71-79, 82-88
nvtabular/framework_utils/torch/models.py                         41      6     22      7    76%   56, 59, 64, 85, 87-88, 91->94, 98->100
nvtabular/framework_utils/torch/utils.py                          31      7     10      3    76%   52, 56-58, 67-69
nvtabular/inference/__init__.py                                    0      0      0      0   100%
nvtabular/inference/triton/__init__.py                           218    211     90      0     2%   24-494
nvtabular/inference/triton/model.py                               56     56     22      0     0%   27-142
nvtabular/inference/triton/model_hugectr.py                       44     44     14      0     0%   27-120
nvtabular/io/__init__.py                                           4      0      0      0   100%
nvtabular/io/avro.py                                              88     88     30      0     0%   16-189
nvtabular/io/csv.py                                               54      4     20      5    88%   95, 99->103, 104, 106, 120
nvtabular/io/dask.py                                             178      7     68     11    93%   109, 112, 148, 223, 380->378, 408->411, 419, 423->425, 425->421, 430, 432
nvtabular/io/dataframe_engine.py                                  58      3     30      6    90%   44, 63, 82->86, 86->91, 88->91, 91->110, 119
nvtabular/io/dataset.py                                          245     30    112     20    85%   254, 256, 269, 278, 296-310, 409->474, 414-417, 422->432, 427-428, 439->437, 453->457, 503, 604->606, 606->615, 616, 623-624, 630, 636, 731-732, 844-849, 855, 880
nvtabular/io/dataset_engine.py                                    23      1      0      0    96%   45
nvtabular/io/hugectr.py                                           45      2     24      2    91%   34, 74->97, 101
nvtabular/io/parquet.py                                          472     28    150     13    93%   83-91, 136-143, 155, 184-186, 311-316, 354-359, 475->482, 545->550, 551-552, 672, 676, 680, 718, 735, 739, 746->748, 866->871, 876->886, 913
nvtabular/io/shuffle.py                                           30      7     12      2    69%   41-48
nvtabular/io/writer.py                                           169     12     64      5    92%   31, 48, 76, 122, 125, 202, 211, 214, 257, 278-280
nvtabular/io/writer_factory.py                                    18      2      8      2    85%   35, 60
nvtabular/loader/__init__.py                                       0      0      0      0   100%
nvtabular/loader/backend.py                                      278     25    114     11    90%   79, 84-85, 113, 130, 137-138, 142-146, 149, 214-218, 224-228, 238, 261-262, 301->305, 376, 380-381, 475, 482
nvtabular/loader/tensorflow.py                                   120     12     52      8    87%   52, 60-63, 73, 83, 287, 294, 302-304, 314->318, 347
nvtabular/loader/tf_utils.py                                      55     10     20      5    80%   29->32, 32->34, 39->41, 43, 50-51, 58-60, 66-70
nvtabular/loader/torch.py                                         43     10      8      0    69%   25-27, 30-36
nvtabular/ops/__init__.py                                         19      0      0      0   100%
nvtabular/ops/bucketize.py                                        25      4     16      2    76%   46, 49-52
nvtabular/ops/categorify.py                                      513     69    298     47    83%   223, 240, 244, 252, 260, 262, 284, 303-304, 336-337, 399-401, 459-461, 466->468, 541, 579, 608->611, 612-614, 621-622, 635-637, 638->606, 654, 664, 666, 672, 688-689, 694, 697->700, 710, 734-736, 739, 741->743, 755-758, 784, 788, 790, 802-805, 920, 922, 964->985, 970->985, 986-991, 1028, 1044->1049, 1048, 1058->1055, 1063->1055, 1071, 1079-1089
nvtabular/ops/clip.py                                             19      2      6      3    80%   45, 53->55, 56
nvtabular/ops/column_similarity.py                                88     22     32      5    69%   84, 154-155, 164-166, 174-190, 205->215, 207->210, 211, 221
nvtabular/ops/data_stats.py                                       57      1     24      4    94%   88->90, 90->92, 93->84, 96->84, 101
nvtabular/ops/difference_lag.py                                   26      0      8      1    97%   67->69
nvtabular/ops/dropna.py                                            9      0      0      0   100%
nvtabular/ops/fill.py                                             58      2     20      1    96%   93, 119
nvtabular/ops/filter.py                                           21      1      6      1    93%   44
nvtabular/ops/hash_bucket.py                                      32      2     18      2    88%   73, 102
nvtabular/ops/hashed_cross.py                                     29      3     13      4    83%   51, 66, 80->exit, 81
nvtabular/ops/join_external.py                                    69      5     28      6    89%   96, 98, 116, 129->133, 153, 168
nvtabular/ops/join_groupby.py                                     80      5     28      2    94%   106, 109->116, 183-184, 187-188
nvtabular/ops/lambdaop.py                                         27      3     10      3    84%   61, 65, 78
nvtabular/ops/logop.py                                             9      0      0      0   100%
nvtabular/ops/moments.py                                          65      0     20      0   100%
nvtabular/ops/normalize.py                                        64      6     14      2    87%   61->60, 67-68, 100-101, 123-124
nvtabular/ops/operator.py                                         15      1      2      1    88%   24
nvtabular/ops/rename.py                                           18      3     10      3    71%   41, 54, 58
nvtabular/ops/stat_operator.py                                    11      0      0      0   100%
nvtabular/ops/target_encoding.py                                 151     11     66      5    91%   143, 163->167, 170->179, 222-223, 226-227, 236-242, 335->338
nvtabular/tools/__init__.py                                        0      0      0      0   100%
nvtabular/tools/data_gen.py                                      235      1     62      4    98%   183->185, 230->234, 320->319, 322
nvtabular/tools/dataset_inspector.py                              49      9     18      0    75%   29-38
nvtabular/tools/inspector_script.py                               46     46      0      0     0%   17-168
nvtabular/utils.py                                                42     16     18      7    55%   20-21, 25-26, 35, 39, 48-51, 53-55, 58, 61, 67, 73
nvtabular/worker.py                                               68      1     30      2    97%   73, 83->98
nvtabular/workflow.py                                            140      9     62      4    93%   39, 125, 137-139, 242, 270-271, 341
----------------------------------------------------------------------------------------------------------
TOTAL                                                           4907   1028   2024    228    76%
Coverage XML written to file coverage.xml

Required test coverage of 70% reached. Total coverage: 76.31%
=========================== short test summary info ============================
FAILED tests/unit/test_tf_dataloader.py::test_tf_gpu_dl[True-1-parquet-0.06]
FAILED tests/unit/test_tf_dataloader.py::test_tf_gpu_dl[True-10-parquet-0.01]
FAILED tests/unit/test_tf_dataloader.py::test_tf_gpu_dl[True-10-parquet-0.06]
FAILED tests/unit/test_tf_dataloader.py::test_tf_gpu_dl[True-100-parquet-0.01]
FAILED tests/unit/test_tf_dataloader.py::test_tf_gpu_dl[True-100-parquet-0.06]
FAILED tests/unit/test_tf_dataloader.py::test_tf_gpu_dl[False-1-parquet-0.01]
FAILED tests/unit/test_tf_dataloader.py::test_tf_gpu_dl[False-1-parquet-0.06]
FAILED tests/unit/test_tf_dataloader.py::test_tf_gpu_dl[False-10-parquet-0.01]
FAILED tests/unit/test_tf_dataloader.py::test_tf_gpu_dl[False-10-parquet-0.06]
FAILED tests/unit/test_tf_dataloader.py::test_tf_gpu_dl[False-100-parquet-0.01]
FAILED tests/unit/test_tf_dataloader.py::test_tf_gpu_dl[False-100-parquet-0.06]
============ 11 failed, 748 passed, 13 skipped in 587.95s (0:09:47) ============
Build step 'Execute shell' marked build as failure
Performing Post build task...
Match found for : : True
Logical operation result is TRUE
Running script  : #!/bin/bash
source activate rapids
cd /var/jenkins_home/
CUDA_VISIBLE_DEVICES=1 python test_res_push.py "https://api.GitHub.com/repos/NVIDIA/NVTabular/issues/$ghprbPullId/comments" "/var/jenkins_home/jobs/$JOB_NAME/builds/$BUILD_NUMBER/log" 
[nvtabular_tests] $ /bin/bash /tmp/jenkins5032331935929203304.sh
</pre></details>
<details><summary>Click to view CI Results</summary><pre>GitHub pull request #677 of commit 6503e5069dea0f27dff7643a464d336bb1d752a8, no merge conflicts.
Running as SYSTEM
Setting status of 6503e5069dea0f27dff7643a464d336bb1d752a8 to PENDING with url http://10.20.13.93:8080/job/nvtabular_tests/2011/ and message: 'Pending'
Using context: Jenkins Unit Test Run
Building in workspace /var/jenkins_home/workspace/nvtabular_tests
using credential ghub_token
Cloning the remote Git repository
Cloning repository https://github.com/NVIDIA/NVTabular.git
 > git init /var/jenkins_home/workspace/nvtabular_tests/nvtabular # timeout=10
Fetching upstream changes from https://github.com/NVIDIA/NVTabular.git
 > git --version # timeout=10
using GIT_ASKPASS to set credentials github token setup
 > git fetch --tags --force --progress -- https://github.com/NVIDIA/NVTabular.git +refs/heads/*:refs/remotes/origin/* # timeout=10
 > git config remote.origin.url https://github.com/NVIDIA/NVTabular.git # timeout=10
 > git config --add remote.origin.fetch +refs/heads/*:refs/remotes/origin/* # timeout=10
 > git config remote.origin.url https://github.com/NVIDIA/NVTabular.git # timeout=10
Fetching upstream changes from https://github.com/NVIDIA/NVTabular.git
using GIT_ASKPASS to set credentials github token setup
 > git fetch --tags --force --progress -- https://github.com/NVIDIA/NVTabular.git +refs/pull/677/*:refs/remotes/origin/pr/677/* # timeout=10
 > git rev-parse 6503e5069dea0f27dff7643a464d336bb1d752a8^{commit} # timeout=10
Checking out Revision 6503e5069dea0f27dff7643a464d336bb1d752a8 (detached)
 > git config core.sparsecheckout # timeout=10
 > git checkout -f 6503e5069dea0f27dff7643a464d336bb1d752a8 # timeout=10
Commit message: "Merge branch 'main' into hive-partitioning"
 > git rev-list --no-walk 42f58af7c1b8c1b29f31c482329dbf6bdd410c24 # timeout=10
First time build. Skipping changelog.
[nvtabular_tests] $ /bin/bash /tmp/jenkins5873751992607622620.sh
Obtaining file:///var/jenkins_home/workspace/nvtabular_tests/nvtabular
  Installing build dependencies: started
  Installing build dependencies: finished with status 'done'
  Getting requirements to build wheel: started
  Getting requirements to build wheel: finished with status 'done'
    Preparing wheel metadata: started
    Preparing wheel metadata: finished with status 'done'
Installing collected packages: nvtabular
  Running setup.py develop for nvtabular
Successfully installed nvtabular
All done! ✨ 🍰 ✨
95 files would be left unchanged.
/conda/envs/rapids/lib/python3.8/site-packages/isort/files.py:30: UserWarning: Likely recursive symlink detected to /var/jenkins_home/workspace/nvtabular_tests/nvtabular/images
  warn(f"Likely recursive symlink detected to {resolved_path}")
============================= test session starts ==============================
platform linux -- Python 3.8.8, pytest-6.2.2, py-1.10.0, pluggy-0.13.1
rootdir: /var/jenkins_home/workspace/nvtabular_tests/nvtabular, configfile: pyproject.toml
plugins: cov-2.11.1, xdist-2.2.1, forked-1.3.0
collected 767 items / 2 skipped / 765 selected

tests/unit/test_column_group.py .                                        [  0%]
tests/unit/test_column_similarity.py ......                              [  0%]
tests/unit/test_dask_nvt.py ............................................ [  6%]
.....................................................................    [ 15%]
tests/unit/test_io.py .................................................. [ 22%]
..................................................................ssssss [ 31%]
ss..............................................                         [ 37%]
tests/unit/test_notebooks.py ..s..                                       [ 38%]
tests/unit/test_ops.py ................................................. [ 44%]
........................................................................ [ 54%]
........................................................................ [ 63%]
...................................                                      [ 68%]
tests/unit/test_tf_dataloader.py .FFFFFFFFFFFF............s              [ 71%]
tests/unit/test_tf_layers.py ........................................... [ 77%]
...................................                                      [ 81%]
tests/unit/test_tools.py ......................                          [ 84%]
tests/unit/test_torch_dataloader.py ......FF..FF..FFFFFFFFFF.ss          [ 88%]
tests/unit/test_workflow.py ............................................ [ 93%]
...............................................                          [100%]

=================================== FAILURES ===================================
_____________________ test_tf_gpu_dl[True-1-parquet-0.01] ______________________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-55/test_tf_gpu_dl_True_1_parquet_0')
paths = ['/tmp/pytest-of-jenkins/pytest-55/parquet0/dataset-0.parquet', '/tmp/pytest-of-jenkins/pytest-55/parquet0/dataset-1.parquet']
use_paths = True
dataset = <nvtabular.io.dataset.Dataset object at 0x7ff031327790>
batch_size = 1, gpu_memory_frac = 0.01, engine = 'parquet'

    @pytest.mark.parametrize("gpu_memory_frac", [0.01, 0.06])
    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("batch_size", [1, 10, 100])
    @pytest.mark.parametrize("use_paths", [True, False])
    def test_tf_gpu_dl(tmpdir, paths, use_paths, dataset, batch_size, gpu_memory_frac, engine):
        cont_names = ["x", "y", "id"]
        cat_names = ["name-string"]
        label_name = ["label"]
        if engine == "parquet":
            cat_names.append("name-cat")
    
        columns = cont_names + cat_names
    
        conts = cont_names >> ops.FillMedian() >> ops.Normalize()
        cats = cat_names >> ops.Categorify()
    
        workflow = nvt.Workflow(conts + cats + label_name)
>       workflow.fit(dataset)

tests/unit/test_tf_dataloader.py:91: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:144: in fit
    results = dask.compute(stats, scheduler="synchronous")[0]
/conda/envs/rapids/lib/python3.8/site-packages/dask/base.py:563: in compute
    results = schedule(dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:528: in get_sync
    return get_async(apply_sync, 1, dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:495: in get_async
    fire_task()
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:457: in fire_task
    apply_async(
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:517: in apply_sync
    res = func(*args, **kwds)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:227: in execute_task
    result = pack_exception(e, dumps)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:222: in execute_task
    result = _execute_task(task, data)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/optimization.py:963: in __call__
    return core.get(self.dsk, self.outkey, dict(zip(self.inkeys, args)))
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:151: in get
    result = _execute_task(task, cache)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/utils.py:35: in apply
    return func(*args, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/dataframe/core.py:5471: in apply_and_enforce
    df = func(*args, **kwargs)
nvtabular/workflow.py:336: in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
nvtabular/ops/fill.py:98: in transform
    df[col] = df[col].fillna(self.medians[col])
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py:1874: in fillna
    return super().fillna(
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py:1475: in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <cudf.core.column.numerical.NumericalColumn object at 0x7ff0312d2540>
fill_value = 1000.5, method = None, dtype = None, fill_nan = True

    def fillna(
        self,
        fill_value: Any = None,
        method: str = None,
        dtype: Dtype = None,
        fill_nan: bool = True,
    ) -> NumericalColumn:
        """
        Fill null values with *fill_value*
        """
        if fill_nan:
            col = self.nans_to_nulls()
        else:
            col = self
    
        if method is not None:
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if fill_value is None:
            raise ValueError("Must specify either 'fill_value' or 'method'")
    
        if (
            isinstance(fill_value, cudf.Scalar)
            and fill_value.dtype == col.dtype
        ):
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if np.isscalar(fill_value):
            # cast safely to the same dtype as self
            fill_value_casted = col.dtype.type(fill_value)
            if not np.isnan(fill_value) and (fill_value_casted != fill_value):
>               raise TypeError(
                    f"Cannot safely cast non-equivalent "
                    f"{type(fill_value).__name__} to {col.dtype.name}"
                )
E               TypeError: Cannot safely cast non-equivalent float to int64

/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py:487: TypeError
------------------------------ Captured log call -------------------------------
ERROR    nvtabular:workflow.py:338 Failed to transform operator <nvtabular.ops.fill.FillMedian object at 0x7ff033a150d0>
Traceback (most recent call last):
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/workflow.py", line 336, in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
  File "/conda/envs/rapids/lib/python3.8/contextlib.py", line 75, in inner
    return func(*args, **kwds)
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/ops/fill.py", line 98, in transform
    df[col] = df[col].fillna(self.medians[col])
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py", line 1874, in fillna
    return super().fillna(
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py", line 1475, in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py", line 487, in fillna
    raise TypeError(
TypeError: Cannot safely cast non-equivalent float to int64
_____________________ test_tf_gpu_dl[True-1-parquet-0.06] ______________________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-55/test_tf_gpu_dl_True_1_parquet_1')
paths = ['/tmp/pytest-of-jenkins/pytest-55/parquet0/dataset-0.parquet', '/tmp/pytest-of-jenkins/pytest-55/parquet0/dataset-1.parquet']
use_paths = True
dataset = <nvtabular.io.dataset.Dataset object at 0x7fef78727a90>
batch_size = 1, gpu_memory_frac = 0.06, engine = 'parquet'

    @pytest.mark.parametrize("gpu_memory_frac", [0.01, 0.06])
    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("batch_size", [1, 10, 100])
    @pytest.mark.parametrize("use_paths", [True, False])
    def test_tf_gpu_dl(tmpdir, paths, use_paths, dataset, batch_size, gpu_memory_frac, engine):
        cont_names = ["x", "y", "id"]
        cat_names = ["name-string"]
        label_name = ["label"]
        if engine == "parquet":
            cat_names.append("name-cat")
    
        columns = cont_names + cat_names
    
        conts = cont_names >> ops.FillMedian() >> ops.Normalize()
        cats = cat_names >> ops.Categorify()
    
        workflow = nvt.Workflow(conts + cats + label_name)
>       workflow.fit(dataset)

tests/unit/test_tf_dataloader.py:91: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:144: in fit
    results = dask.compute(stats, scheduler="synchronous")[0]
/conda/envs/rapids/lib/python3.8/site-packages/dask/base.py:563: in compute
    results = schedule(dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:528: in get_sync
    return get_async(apply_sync, 1, dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:495: in get_async
    fire_task()
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:457: in fire_task
    apply_async(
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:517: in apply_sync
    res = func(*args, **kwds)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:227: in execute_task
    result = pack_exception(e, dumps)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:222: in execute_task
    result = _execute_task(task, data)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/optimization.py:963: in __call__
    return core.get(self.dsk, self.outkey, dict(zip(self.inkeys, args)))
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:151: in get
    result = _execute_task(task, cache)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/utils.py:35: in apply
    return func(*args, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/dataframe/core.py:5471: in apply_and_enforce
    df = func(*args, **kwargs)
nvtabular/workflow.py:336: in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
nvtabular/ops/fill.py:98: in transform
    df[col] = df[col].fillna(self.medians[col])
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py:1874: in fillna
    return super().fillna(
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py:1475: in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <cudf.core.column.numerical.NumericalColumn object at 0x7ff0286909c0>
fill_value = 1000.5, method = None, dtype = None, fill_nan = True

    def fillna(
        self,
        fill_value: Any = None,
        method: str = None,
        dtype: Dtype = None,
        fill_nan: bool = True,
    ) -> NumericalColumn:
        """
        Fill null values with *fill_value*
        """
        if fill_nan:
            col = self.nans_to_nulls()
        else:
            col = self
    
        if method is not None:
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if fill_value is None:
            raise ValueError("Must specify either 'fill_value' or 'method'")
    
        if (
            isinstance(fill_value, cudf.Scalar)
            and fill_value.dtype == col.dtype
        ):
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if np.isscalar(fill_value):
            # cast safely to the same dtype as self
            fill_value_casted = col.dtype.type(fill_value)
            if not np.isnan(fill_value) and (fill_value_casted != fill_value):
>               raise TypeError(
                    f"Cannot safely cast non-equivalent "
                    f"{type(fill_value).__name__} to {col.dtype.name}"
                )
E               TypeError: Cannot safely cast non-equivalent float to int64

/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py:487: TypeError
------------------------------ Captured log call -------------------------------
ERROR    nvtabular:workflow.py:338 Failed to transform operator <nvtabular.ops.fill.FillMedian object at 0x7fef78727ee0>
Traceback (most recent call last):
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/workflow.py", line 336, in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
  File "/conda/envs/rapids/lib/python3.8/contextlib.py", line 75, in inner
    return func(*args, **kwds)
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/ops/fill.py", line 98, in transform
    df[col] = df[col].fillna(self.medians[col])
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py", line 1874, in fillna
    return super().fillna(
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py", line 1475, in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py", line 487, in fillna
    raise TypeError(
TypeError: Cannot safely cast non-equivalent float to int64
_____________________ test_tf_gpu_dl[True-10-parquet-0.01] _____________________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-55/test_tf_gpu_dl_True_10_parquet0')
paths = ['/tmp/pytest-of-jenkins/pytest-55/parquet0/dataset-0.parquet', '/tmp/pytest-of-jenkins/pytest-55/parquet0/dataset-1.parquet']
use_paths = True
dataset = <nvtabular.io.dataset.Dataset object at 0x7fef787a6e80>
batch_size = 10, gpu_memory_frac = 0.01, engine = 'parquet'

    @pytest.mark.parametrize("gpu_memory_frac", [0.01, 0.06])
    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("batch_size", [1, 10, 100])
    @pytest.mark.parametrize("use_paths", [True, False])
    def test_tf_gpu_dl(tmpdir, paths, use_paths, dataset, batch_size, gpu_memory_frac, engine):
        cont_names = ["x", "y", "id"]
        cat_names = ["name-string"]
        label_name = ["label"]
        if engine == "parquet":
            cat_names.append("name-cat")
    
        columns = cont_names + cat_names
    
        conts = cont_names >> ops.FillMedian() >> ops.Normalize()
        cats = cat_names >> ops.Categorify()
    
        workflow = nvt.Workflow(conts + cats + label_name)
>       workflow.fit(dataset)

tests/unit/test_tf_dataloader.py:91: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:144: in fit
    results = dask.compute(stats, scheduler="synchronous")[0]
/conda/envs/rapids/lib/python3.8/site-packages/dask/base.py:563: in compute
    results = schedule(dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:528: in get_sync
    return get_async(apply_sync, 1, dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:495: in get_async
    fire_task()
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:457: in fire_task
    apply_async(
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:517: in apply_sync
    res = func(*args, **kwds)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:227: in execute_task
    result = pack_exception(e, dumps)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:222: in execute_task
    result = _execute_task(task, data)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/optimization.py:963: in __call__
    return core.get(self.dsk, self.outkey, dict(zip(self.inkeys, args)))
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:151: in get
    result = _execute_task(task, cache)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/utils.py:35: in apply
    return func(*args, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/dataframe/core.py:5471: in apply_and_enforce
    df = func(*args, **kwargs)
nvtabular/workflow.py:336: in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
nvtabular/ops/fill.py:98: in transform
    df[col] = df[col].fillna(self.medians[col])
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py:1874: in fillna
    return super().fillna(
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py:1475: in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <cudf.core.column.numerical.NumericalColumn object at 0x7ff0313a5c40>
fill_value = 1000.5, method = None, dtype = None, fill_nan = True

    def fillna(
        self,
        fill_value: Any = None,
        method: str = None,
        dtype: Dtype = None,
        fill_nan: bool = True,
    ) -> NumericalColumn:
        """
        Fill null values with *fill_value*
        """
        if fill_nan:
            col = self.nans_to_nulls()
        else:
            col = self
    
        if method is not None:
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if fill_value is None:
            raise ValueError("Must specify either 'fill_value' or 'method'")
    
        if (
            isinstance(fill_value, cudf.Scalar)
            and fill_value.dtype == col.dtype
        ):
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if np.isscalar(fill_value):
            # cast safely to the same dtype as self
            fill_value_casted = col.dtype.type(fill_value)
            if not np.isnan(fill_value) and (fill_value_casted != fill_value):
>               raise TypeError(
                    f"Cannot safely cast non-equivalent "
                    f"{type(fill_value).__name__} to {col.dtype.name}"
                )
E               TypeError: Cannot safely cast non-equivalent float to int64

/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py:487: TypeError
------------------------------ Captured log call -------------------------------
ERROR    nvtabular:workflow.py:338 Failed to transform operator <nvtabular.ops.fill.FillMedian object at 0x7fef787a62b0>
Traceback (most recent call last):
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/workflow.py", line 336, in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
  File "/conda/envs/rapids/lib/python3.8/contextlib.py", line 75, in inner
    return func(*args, **kwds)
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/ops/fill.py", line 98, in transform
    df[col] = df[col].fillna(self.medians[col])
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py", line 1874, in fillna
    return super().fillna(
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py", line 1475, in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py", line 487, in fillna
    raise TypeError(
TypeError: Cannot safely cast non-equivalent float to int64
_____________________ test_tf_gpu_dl[True-10-parquet-0.06] _____________________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-55/test_tf_gpu_dl_True_10_parquet1')
paths = ['/tmp/pytest-of-jenkins/pytest-55/parquet0/dataset-0.parquet', '/tmp/pytest-of-jenkins/pytest-55/parquet0/dataset-1.parquet']
use_paths = True
dataset = <nvtabular.io.dataset.Dataset object at 0x7ff028119880>
batch_size = 10, gpu_memory_frac = 0.06, engine = 'parquet'

    @pytest.mark.parametrize("gpu_memory_frac", [0.01, 0.06])
    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("batch_size", [1, 10, 100])
    @pytest.mark.parametrize("use_paths", [True, False])
    def test_tf_gpu_dl(tmpdir, paths, use_paths, dataset, batch_size, gpu_memory_frac, engine):
        cont_names = ["x", "y", "id"]
        cat_names = ["name-string"]
        label_name = ["label"]
        if engine == "parquet":
            cat_names.append("name-cat")
    
        columns = cont_names + cat_names
    
        conts = cont_names >> ops.FillMedian() >> ops.Normalize()
        cats = cat_names >> ops.Categorify()
    
        workflow = nvt.Workflow(conts + cats + label_name)
>       workflow.fit(dataset)

tests/unit/test_tf_dataloader.py:91: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:144: in fit
    results = dask.compute(stats, scheduler="synchronous")[0]
/conda/envs/rapids/lib/python3.8/site-packages/dask/base.py:563: in compute
    results = schedule(dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:528: in get_sync
    return get_async(apply_sync, 1, dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:495: in get_async
    fire_task()
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:457: in fire_task
    apply_async(
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:517: in apply_sync
    res = func(*args, **kwds)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:227: in execute_task
    result = pack_exception(e, dumps)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:222: in execute_task
    result = _execute_task(task, data)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/optimization.py:963: in __call__
    return core.get(self.dsk, self.outkey, dict(zip(self.inkeys, args)))
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:151: in get
    result = _execute_task(task, cache)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/utils.py:35: in apply
    return func(*args, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/dataframe/core.py:5471: in apply_and_enforce
    df = func(*args, **kwargs)
nvtabular/workflow.py:336: in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
nvtabular/ops/fill.py:98: in transform
    df[col] = df[col].fillna(self.medians[col])
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py:1874: in fillna
    return super().fillna(
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py:1475: in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <cudf.core.column.numerical.NumericalColumn object at 0x7ff031361840>
fill_value = 1000.5, method = None, dtype = None, fill_nan = True

    def fillna(
        self,
        fill_value: Any = None,
        method: str = None,
        dtype: Dtype = None,
        fill_nan: bool = True,
    ) -> NumericalColumn:
        """
        Fill null values with *fill_value*
        """
        if fill_nan:
            col = self.nans_to_nulls()
        else:
            col = self
    
        if method is not None:
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if fill_value is None:
            raise ValueError("Must specify either 'fill_value' or 'method'")
    
        if (
            isinstance(fill_value, cudf.Scalar)
            and fill_value.dtype == col.dtype
        ):
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if np.isscalar(fill_value):
            # cast safely to the same dtype as self
            fill_value_casted = col.dtype.type(fill_value)
            if not np.isnan(fill_value) and (fill_value_casted != fill_value):
>               raise TypeError(
                    f"Cannot safely cast non-equivalent "
                    f"{type(fill_value).__name__} to {col.dtype.name}"
                )
E               TypeError: Cannot safely cast non-equivalent float to int64

/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py:487: TypeError
------------------------------ Captured log call -------------------------------
ERROR    nvtabular:workflow.py:338 Failed to transform operator <nvtabular.ops.fill.FillMedian object at 0x7ff028119a00>
Traceback (most recent call last):
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/workflow.py", line 336, in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
  File "/conda/envs/rapids/lib/python3.8/contextlib.py", line 75, in inner
    return func(*args, **kwds)
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/ops/fill.py", line 98, in transform
    df[col] = df[col].fillna(self.medians[col])
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py", line 1874, in fillna
    return super().fillna(
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py", line 1475, in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py", line 487, in fillna
    raise TypeError(
TypeError: Cannot safely cast non-equivalent float to int64
____________________ test_tf_gpu_dl[True-100-parquet-0.01] _____________________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-55/test_tf_gpu_dl_True_100_parque0')
paths = ['/tmp/pytest-of-jenkins/pytest-55/parquet0/dataset-0.parquet', '/tmp/pytest-of-jenkins/pytest-55/parquet0/dataset-1.parquet']
use_paths = True
dataset = <nvtabular.io.dataset.Dataset object at 0x7ff018374580>
batch_size = 100, gpu_memory_frac = 0.01, engine = 'parquet'

    @pytest.mark.parametrize("gpu_memory_frac", [0.01, 0.06])
    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("batch_size", [1, 10, 100])
    @pytest.mark.parametrize("use_paths", [True, False])
    def test_tf_gpu_dl(tmpdir, paths, use_paths, dataset, batch_size, gpu_memory_frac, engine):
        cont_names = ["x", "y", "id"]
        cat_names = ["name-string"]
        label_name = ["label"]
        if engine == "parquet":
            cat_names.append("name-cat")
    
        columns = cont_names + cat_names
    
        conts = cont_names >> ops.FillMedian() >> ops.Normalize()
        cats = cat_names >> ops.Categorify()
    
        workflow = nvt.Workflow(conts + cats + label_name)
>       workflow.fit(dataset)

tests/unit/test_tf_dataloader.py:91: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:144: in fit
    results = dask.compute(stats, scheduler="synchronous")[0]
/conda/envs/rapids/lib/python3.8/site-packages/dask/base.py:563: in compute
    results = schedule(dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:528: in get_sync
    return get_async(apply_sync, 1, dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:495: in get_async
    fire_task()
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:457: in fire_task
    apply_async(
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:517: in apply_sync
    res = func(*args, **kwds)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:227: in execute_task
    result = pack_exception(e, dumps)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:222: in execute_task
    result = _execute_task(task, data)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/optimization.py:963: in __call__
    return core.get(self.dsk, self.outkey, dict(zip(self.inkeys, args)))
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:151: in get
    result = _execute_task(task, cache)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/utils.py:35: in apply
    return func(*args, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/dataframe/core.py:5471: in apply_and_enforce
    df = func(*args, **kwargs)
nvtabular/workflow.py:336: in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
nvtabular/ops/fill.py:98: in transform
    df[col] = df[col].fillna(self.medians[col])
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py:1874: in fillna
    return super().fillna(
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py:1475: in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <cudf.core.column.numerical.NumericalColumn object at 0x7ff0402afec0>
fill_value = 1000.5, method = None, dtype = None, fill_nan = True

    def fillna(
        self,
        fill_value: Any = None,
        method: str = None,
        dtype: Dtype = None,
        fill_nan: bool = True,
    ) -> NumericalColumn:
        """
        Fill null values with *fill_value*
        """
        if fill_nan:
            col = self.nans_to_nulls()
        else:
            col = self
    
        if method is not None:
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if fill_value is None:
            raise ValueError("Must specify either 'fill_value' or 'method'")
    
        if (
            isinstance(fill_value, cudf.Scalar)
            and fill_value.dtype == col.dtype
        ):
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if np.isscalar(fill_value):
            # cast safely to the same dtype as self
            fill_value_casted = col.dtype.type(fill_value)
            if not np.isnan(fill_value) and (fill_value_casted != fill_value):
>               raise TypeError(
                    f"Cannot safely cast non-equivalent "
                    f"{type(fill_value).__name__} to {col.dtype.name}"
                )
E               TypeError: Cannot safely cast non-equivalent float to int64

/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py:487: TypeError
------------------------------ Captured log call -------------------------------
ERROR    nvtabular:workflow.py:338 Failed to transform operator <nvtabular.ops.fill.FillMedian object at 0x7ff018374a00>
Traceback (most recent call last):
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/workflow.py", line 336, in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
  File "/conda/envs/rapids/lib/python3.8/contextlib.py", line 75, in inner
    return func(*args, **kwds)
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/ops/fill.py", line 98, in transform
    df[col] = df[col].fillna(self.medians[col])
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py", line 1874, in fillna
    return super().fillna(
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py", line 1475, in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py", line 487, in fillna
    raise TypeError(
TypeError: Cannot safely cast non-equivalent float to int64
____________________ test_tf_gpu_dl[True-100-parquet-0.06] _____________________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-55/test_tf_gpu_dl_True_100_parque1')
paths = ['/tmp/pytest-of-jenkins/pytest-55/parquet0/dataset-0.parquet', '/tmp/pytest-of-jenkins/pytest-55/parquet0/dataset-1.parquet']
use_paths = True
dataset = <nvtabular.io.dataset.Dataset object at 0x7fef782ef970>
batch_size = 100, gpu_memory_frac = 0.06, engine = 'parquet'

    @pytest.mark.parametrize("gpu_memory_frac", [0.01, 0.06])
    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("batch_size", [1, 10, 100])
    @pytest.mark.parametrize("use_paths", [True, False])
    def test_tf_gpu_dl(tmpdir, paths, use_paths, dataset, batch_size, gpu_memory_frac, engine):
        cont_names = ["x", "y", "id"]
        cat_names = ["name-string"]
        label_name = ["label"]
        if engine == "parquet":
            cat_names.append("name-cat")
    
        columns = cont_names + cat_names
    
        conts = cont_names >> ops.FillMedian() >> ops.Normalize()
        cats = cat_names >> ops.Categorify()
    
        workflow = nvt.Workflow(conts + cats + label_name)
>       workflow.fit(dataset)

tests/unit/test_tf_dataloader.py:91: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:147: in fit
    op.fit_finalize(computed_stats)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <nvtabular.ops.fill.FillMedian object at 0x7fef782efdc0>
dask_stats = x     0.017761632
y            <NA>
id         1000.5
Name: 0.5, dtype: float64

    @annotate("FillMedian_finalize", color="green", domain="nvt_python")
    def fit_finalize(self, dask_stats):
        index = dask_stats.index
        vals = index.values_host if hasattr(index, "values_host") else index.values
        for col in vals:
>           self.medians[col] = float(dask_stats[col])
E           TypeError: float() argument must be a string or a number, not '_NAType'

nvtabular/ops/fill.py:112: TypeError
_____________________ test_tf_gpu_dl[False-1-parquet-0.01] _____________________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-55/test_tf_gpu_dl_False_1_parquet0')
paths = ['/tmp/pytest-of-jenkins/pytest-55/parquet0/dataset-0.parquet', '/tmp/pytest-of-jenkins/pytest-55/parquet0/dataset-1.parquet']
use_paths = False
dataset = <nvtabular.io.dataset.Dataset object at 0x7ff0f60ad790>
batch_size = 1, gpu_memory_frac = 0.01, engine = 'parquet'

    @pytest.mark.parametrize("gpu_memory_frac", [0.01, 0.06])
    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("batch_size", [1, 10, 100])
    @pytest.mark.parametrize("use_paths", [True, False])
    def test_tf_gpu_dl(tmpdir, paths, use_paths, dataset, batch_size, gpu_memory_frac, engine):
        cont_names = ["x", "y", "id"]
        cat_names = ["name-string"]
        label_name = ["label"]
        if engine == "parquet":
            cat_names.append("name-cat")
    
        columns = cont_names + cat_names
    
        conts = cont_names >> ops.FillMedian() >> ops.Normalize()
        cats = cat_names >> ops.Categorify()
    
        workflow = nvt.Workflow(conts + cats + label_name)
>       workflow.fit(dataset)

tests/unit/test_tf_dataloader.py:91: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:144: in fit
    results = dask.compute(stats, scheduler="synchronous")[0]
/conda/envs/rapids/lib/python3.8/site-packages/dask/base.py:563: in compute
    results = schedule(dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:528: in get_sync
    return get_async(apply_sync, 1, dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:495: in get_async
    fire_task()
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:457: in fire_task
    apply_async(
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:517: in apply_sync
    res = func(*args, **kwds)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:227: in execute_task
    result = pack_exception(e, dumps)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:222: in execute_task
    result = _execute_task(task, data)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/optimization.py:963: in __call__
    return core.get(self.dsk, self.outkey, dict(zip(self.inkeys, args)))
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:151: in get
    result = _execute_task(task, cache)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/utils.py:35: in apply
    return func(*args, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/dataframe/core.py:5471: in apply_and_enforce
    df = func(*args, **kwargs)
nvtabular/workflow.py:336: in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
nvtabular/ops/fill.py:98: in transform
    df[col] = df[col].fillna(self.medians[col])
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py:1874: in fillna
    return super().fillna(
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py:1475: in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <cudf.core.column.numerical.NumericalColumn object at 0x7ff03825e6c0>
fill_value = 1000.5, method = None, dtype = None, fill_nan = True

    def fillna(
        self,
        fill_value: Any = None,
        method: str = None,
        dtype: Dtype = None,
        fill_nan: bool = True,
    ) -> NumericalColumn:
        """
        Fill null values with *fill_value*
        """
        if fill_nan:
            col = self.nans_to_nulls()
        else:
            col = self
    
        if method is not None:
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if fill_value is None:
            raise ValueError("Must specify either 'fill_value' or 'method'")
    
        if (
            isinstance(fill_value, cudf.Scalar)
            and fill_value.dtype == col.dtype
        ):
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if np.isscalar(fill_value):
            # cast safely to the same dtype as self
            fill_value_casted = col.dtype.type(fill_value)
            if not np.isnan(fill_value) and (fill_value_casted != fill_value):
>               raise TypeError(
                    f"Cannot safely cast non-equivalent "
                    f"{type(fill_value).__name__} to {col.dtype.name}"
                )
E               TypeError: Cannot safely cast non-equivalent float to int64

/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py:487: TypeError
------------------------------ Captured log call -------------------------------
ERROR    nvtabular:workflow.py:338 Failed to transform operator <nvtabular.ops.fill.FillMedian object at 0x7ff0f60ad130>
Traceback (most recent call last):
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/workflow.py", line 336, in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
  File "/conda/envs/rapids/lib/python3.8/contextlib.py", line 75, in inner
    return func(*args, **kwds)
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/ops/fill.py", line 98, in transform
    df[col] = df[col].fillna(self.medians[col])
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py", line 1874, in fillna
    return super().fillna(
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py", line 1475, in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py", line 487, in fillna
    raise TypeError(
TypeError: Cannot safely cast non-equivalent float to int64
_____________________ test_tf_gpu_dl[False-1-parquet-0.06] _____________________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-55/test_tf_gpu_dl_False_1_parquet1')
paths = ['/tmp/pytest-of-jenkins/pytest-55/parquet0/dataset-0.parquet', '/tmp/pytest-of-jenkins/pytest-55/parquet0/dataset-1.parquet']
use_paths = False
dataset = <nvtabular.io.dataset.Dataset object at 0x7ff0185432e0>
batch_size = 1, gpu_memory_frac = 0.06, engine = 'parquet'

    @pytest.mark.parametrize("gpu_memory_frac", [0.01, 0.06])
    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("batch_size", [1, 10, 100])
    @pytest.mark.parametrize("use_paths", [True, False])
    def test_tf_gpu_dl(tmpdir, paths, use_paths, dataset, batch_size, gpu_memory_frac, engine):
        cont_names = ["x", "y", "id"]
        cat_names = ["name-string"]
        label_name = ["label"]
        if engine == "parquet":
            cat_names.append("name-cat")
    
        columns = cont_names + cat_names
    
        conts = cont_names >> ops.FillMedian() >> ops.Normalize()
        cats = cat_names >> ops.Categorify()
    
        workflow = nvt.Workflow(conts + cats + label_name)
>       workflow.fit(dataset)

tests/unit/test_tf_dataloader.py:91: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:144: in fit
    results = dask.compute(stats, scheduler="synchronous")[0]
/conda/envs/rapids/lib/python3.8/site-packages/dask/base.py:563: in compute
    results = schedule(dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:528: in get_sync
    return get_async(apply_sync, 1, dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:495: in get_async
    fire_task()
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:457: in fire_task
    apply_async(
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:517: in apply_sync
    res = func(*args, **kwds)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:227: in execute_task
    result = pack_exception(e, dumps)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:222: in execute_task
    result = _execute_task(task, data)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/optimization.py:963: in __call__
    return core.get(self.dsk, self.outkey, dict(zip(self.inkeys, args)))
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:151: in get
    result = _execute_task(task, cache)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/utils.py:35: in apply
    return func(*args, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/dataframe/core.py:5471: in apply_and_enforce
    df = func(*args, **kwargs)
nvtabular/workflow.py:336: in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
nvtabular/ops/fill.py:98: in transform
    df[col] = df[col].fillna(self.medians[col])
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py:1874: in fillna
    return super().fillna(
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py:1475: in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <cudf.core.column.numerical.NumericalColumn object at 0x7ff0313a5f40>
fill_value = 1000.5, method = None, dtype = None, fill_nan = True

    def fillna(
        self,
        fill_value: Any = None,
        method: str = None,
        dtype: Dtype = None,
        fill_nan: bool = True,
    ) -> NumericalColumn:
        """
        Fill null values with *fill_value*
        """
        if fill_nan:
            col = self.nans_to_nulls()
        else:
            col = self
    
        if method is not None:
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if fill_value is None:
            raise ValueError("Must specify either 'fill_value' or 'method'")
    
        if (
            isinstance(fill_value, cudf.Scalar)
            and fill_value.dtype == col.dtype
        ):
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if np.isscalar(fill_value):
            # cast safely to the same dtype as self
            fill_value_casted = col.dtype.type(fill_value)
            if not np.isnan(fill_value) and (fill_value_casted != fill_value):
>               raise TypeError(
                    f"Cannot safely cast non-equivalent "
                    f"{type(fill_value).__name__} to {col.dtype.name}"
                )
E               TypeError: Cannot safely cast non-equivalent float to int64

/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py:487: TypeError
------------------------------ Captured log call -------------------------------
ERROR    nvtabular:workflow.py:338 Failed to transform operator <nvtabular.ops.fill.FillMedian object at 0x7ff018543790>
Traceback (most recent call last):
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/workflow.py", line 336, in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
  File "/conda/envs/rapids/lib/python3.8/contextlib.py", line 75, in inner
    return func(*args, **kwds)
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/ops/fill.py", line 98, in transform
    df[col] = df[col].fillna(self.medians[col])
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py", line 1874, in fillna
    return super().fillna(
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py", line 1475, in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py", line 487, in fillna
    raise TypeError(
TypeError: Cannot safely cast non-equivalent float to int64
____________________ test_tf_gpu_dl[False-10-parquet-0.01] _____________________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-55/test_tf_gpu_dl_False_10_parque0')
paths = ['/tmp/pytest-of-jenkins/pytest-55/parquet0/dataset-0.parquet', '/tmp/pytest-of-jenkins/pytest-55/parquet0/dataset-1.parquet']
use_paths = False
dataset = <nvtabular.io.dataset.Dataset object at 0x7ff018543400>
batch_size = 10, gpu_memory_frac = 0.01, engine = 'parquet'

    @pytest.mark.parametrize("gpu_memory_frac", [0.01, 0.06])
    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("batch_size", [1, 10, 100])
    @pytest.mark.parametrize("use_paths", [True, False])
    def test_tf_gpu_dl(tmpdir, paths, use_paths, dataset, batch_size, gpu_memory_frac, engine):
        cont_names = ["x", "y", "id"]
        cat_names = ["name-string"]
        label_name = ["label"]
        if engine == "parquet":
            cat_names.append("name-cat")
    
        columns = cont_names + cat_names
    
        conts = cont_names >> ops.FillMedian() >> ops.Normalize()
        cats = cat_names >> ops.Categorify()
    
        workflow = nvt.Workflow(conts + cats + label_name)
>       workflow.fit(dataset)

tests/unit/test_tf_dataloader.py:91: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:144: in fit
    results = dask.compute(stats, scheduler="synchronous")[0]
/conda/envs/rapids/lib/python3.8/site-packages/dask/base.py:563: in compute
    results = schedule(dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:528: in get_sync
    return get_async(apply_sync, 1, dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:495: in get_async
    fire_task()
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:457: in fire_task
    apply_async(
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:517: in apply_sync
    res = func(*args, **kwds)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:227: in execute_task
    result = pack_exception(e, dumps)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:222: in execute_task
    result = _execute_task(task, data)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/optimization.py:963: in __call__
    return core.get(self.dsk, self.outkey, dict(zip(self.inkeys, args)))
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:151: in get
    result = _execute_task(task, cache)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/utils.py:35: in apply
    return func(*args, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/dataframe/core.py:5471: in apply_and_enforce
    df = func(*args, **kwargs)
nvtabular/workflow.py:336: in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
nvtabular/ops/fill.py:98: in transform
    df[col] = df[col].fillna(self.medians[col])
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py:1874: in fillna
    return super().fillna(
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py:1475: in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <cudf.core.column.numerical.NumericalColumn object at 0x7ff0402b11c0>
fill_value = 1000.5, method = None, dtype = None, fill_nan = True

    def fillna(
        self,
        fill_value: Any = None,
        method: str = None,
        dtype: Dtype = None,
        fill_nan: bool = True,
    ) -> NumericalColumn:
        """
        Fill null values with *fill_value*
        """
        if fill_nan:
            col = self.nans_to_nulls()
        else:
            col = self
    
        if method is not None:
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if fill_value is None:
            raise ValueError("Must specify either 'fill_value' or 'method'")
    
        if (
            isinstance(fill_value, cudf.Scalar)
            and fill_value.dtype == col.dtype
        ):
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if np.isscalar(fill_value):
            # cast safely to the same dtype as self
            fill_value_casted = col.dtype.type(fill_value)
            if not np.isnan(fill_value) and (fill_value_casted != fill_value):
>               raise TypeError(
                    f"Cannot safely cast non-equivalent "
                    f"{type(fill_value).__name__} to {col.dtype.name}"
                )
E               TypeError: Cannot safely cast non-equivalent float to int64

/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py:487: TypeError
------------------------------ Captured log call -------------------------------
ERROR    nvtabular:workflow.py:338 Failed to transform operator <nvtabular.ops.fill.FillMedian object at 0x7ff0185435b0>
Traceback (most recent call last):
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/workflow.py", line 336, in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
  File "/conda/envs/rapids/lib/python3.8/contextlib.py", line 75, in inner
    return func(*args, **kwds)
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/ops/fill.py", line 98, in transform
    df[col] = df[col].fillna(self.medians[col])
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py", line 1874, in fillna
    return super().fillna(
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py", line 1475, in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py", line 487, in fillna
    raise TypeError(
TypeError: Cannot safely cast non-equivalent float to int64
____________________ test_tf_gpu_dl[False-10-parquet-0.06] _____________________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-55/test_tf_gpu_dl_False_10_parque1')
paths = ['/tmp/pytest-of-jenkins/pytest-55/parquet0/dataset-0.parquet', '/tmp/pytest-of-jenkins/pytest-55/parquet0/dataset-1.parquet']
use_paths = False
dataset = <nvtabular.io.dataset.Dataset object at 0x7ff0280bc910>
batch_size = 10, gpu_memory_frac = 0.06, engine = 'parquet'

    @pytest.mark.parametrize("gpu_memory_frac", [0.01, 0.06])
    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("batch_size", [1, 10, 100])
    @pytest.mark.parametrize("use_paths", [True, False])
    def test_tf_gpu_dl(tmpdir, paths, use_paths, dataset, batch_size, gpu_memory_frac, engine):
        cont_names = ["x", "y", "id"]
        cat_names = ["name-string"]
        label_name = ["label"]
        if engine == "parquet":
            cat_names.append("name-cat")
    
        columns = cont_names + cat_names
    
        conts = cont_names >> ops.FillMedian() >> ops.Normalize()
        cats = cat_names >> ops.Categorify()
    
        workflow = nvt.Workflow(conts + cats + label_name)
>       workflow.fit(dataset)

tests/unit/test_tf_dataloader.py:91: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:144: in fit
    results = dask.compute(stats, scheduler="synchronous")[0]
/conda/envs/rapids/lib/python3.8/site-packages/dask/base.py:563: in compute
    results = schedule(dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:528: in get_sync
    return get_async(apply_sync, 1, dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:495: in get_async
    fire_task()
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:457: in fire_task
    apply_async(
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:517: in apply_sync
    res = func(*args, **kwds)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:227: in execute_task
    result = pack_exception(e, dumps)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:222: in execute_task
    result = _execute_task(task, data)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/optimization.py:963: in __call__
    return core.get(self.dsk, self.outkey, dict(zip(self.inkeys, args)))
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:151: in get
    result = _execute_task(task, cache)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/utils.py:35: in apply
    return func(*args, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/dataframe/core.py:5471: in apply_and_enforce
    df = func(*args, **kwargs)
nvtabular/workflow.py:336: in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
nvtabular/ops/fill.py:98: in transform
    df[col] = df[col].fillna(self.medians[col])
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py:1874: in fillna
    return super().fillna(
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py:1475: in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <cudf.core.column.numerical.NumericalColumn object at 0x7ff0f38faac0>
fill_value = 1000.5, method = None, dtype = None, fill_nan = True

    def fillna(
        self,
        fill_value: Any = None,
        method: str = None,
        dtype: Dtype = None,
        fill_nan: bool = True,
    ) -> NumericalColumn:
        """
        Fill null values with *fill_value*
        """
        if fill_nan:
            col = self.nans_to_nulls()
        else:
            col = self
    
        if method is not None:
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if fill_value is None:
            raise ValueError("Must specify either 'fill_value' or 'method'")
    
        if (
            isinstance(fill_value, cudf.Scalar)
            and fill_value.dtype == col.dtype
        ):
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if np.isscalar(fill_value):
            # cast safely to the same dtype as self
            fill_value_casted = col.dtype.type(fill_value)
            if not np.isnan(fill_value) and (fill_value_casted != fill_value):
>               raise TypeError(
                    f"Cannot safely cast non-equivalent "
                    f"{type(fill_value).__name__} to {col.dtype.name}"
                )
E               TypeError: Cannot safely cast non-equivalent float to int64

/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py:487: TypeError
------------------------------ Captured log call -------------------------------
ERROR    nvtabular:workflow.py:338 Failed to transform operator <nvtabular.ops.fill.FillMedian object at 0x7ff0280bc040>
Traceback (most recent call last):
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/workflow.py", line 336, in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
  File "/conda/envs/rapids/lib/python3.8/contextlib.py", line 75, in inner
    return func(*args, **kwds)
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/ops/fill.py", line 98, in transform
    df[col] = df[col].fillna(self.medians[col])
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py", line 1874, in fillna
    return super().fillna(
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py", line 1475, in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py", line 487, in fillna
    raise TypeError(
TypeError: Cannot safely cast non-equivalent float to int64
____________________ test_tf_gpu_dl[False-100-parquet-0.01] ____________________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-55/test_tf_gpu_dl_False_100_parqu0')
paths = ['/tmp/pytest-of-jenkins/pytest-55/parquet0/dataset-0.parquet', '/tmp/pytest-of-jenkins/pytest-55/parquet0/dataset-1.parquet']
use_paths = False
dataset = <nvtabular.io.dataset.Dataset object at 0x7fef784c2940>
batch_size = 100, gpu_memory_frac = 0.01, engine = 'parquet'

    @pytest.mark.parametrize("gpu_memory_frac", [0.01, 0.06])
    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("batch_size", [1, 10, 100])
    @pytest.mark.parametrize("use_paths", [True, False])
    def test_tf_gpu_dl(tmpdir, paths, use_paths, dataset, batch_size, gpu_memory_frac, engine):
        cont_names = ["x", "y", "id"]
        cat_names = ["name-string"]
        label_name = ["label"]
        if engine == "parquet":
            cat_names.append("name-cat")
    
        columns = cont_names + cat_names
    
        conts = cont_names >> ops.FillMedian() >> ops.Normalize()
        cats = cat_names >> ops.Categorify()
    
        workflow = nvt.Workflow(conts + cats + label_name)
>       workflow.fit(dataset)

tests/unit/test_tf_dataloader.py:91: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:144: in fit
    results = dask.compute(stats, scheduler="synchronous")[0]
/conda/envs/rapids/lib/python3.8/site-packages/dask/base.py:563: in compute
    results = schedule(dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:528: in get_sync
    return get_async(apply_sync, 1, dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:495: in get_async
    fire_task()
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:457: in fire_task
    apply_async(
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:517: in apply_sync
    res = func(*args, **kwds)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:227: in execute_task
    result = pack_exception(e, dumps)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:222: in execute_task
    result = _execute_task(task, data)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/optimization.py:963: in __call__
    return core.get(self.dsk, self.outkey, dict(zip(self.inkeys, args)))
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:151: in get
    result = _execute_task(task, cache)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/utils.py:35: in apply
    return func(*args, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/dataframe/core.py:5471: in apply_and_enforce
    df = func(*args, **kwargs)
nvtabular/workflow.py:336: in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
nvtabular/ops/fill.py:98: in transform
    df[col] = df[col].fillna(self.medians[col])
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py:1874: in fillna
    return super().fillna(
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py:1475: in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <cudf.core.column.numerical.NumericalColumn object at 0x7ff04029aac0>
fill_value = 1000.5, method = None, dtype = None, fill_nan = True

    def fillna(
        self,
        fill_value: Any = None,
        method: str = None,
        dtype: Dtype = None,
        fill_nan: bool = True,
    ) -> NumericalColumn:
        """
        Fill null values with *fill_value*
        """
        if fill_nan:
            col = self.nans_to_nulls()
        else:
            col = self
    
        if method is not None:
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if fill_value is None:
            raise ValueError("Must specify either 'fill_value' or 'method'")
    
        if (
            isinstance(fill_value, cudf.Scalar)
            and fill_value.dtype == col.dtype
        ):
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if np.isscalar(fill_value):
            # cast safely to the same dtype as self
            fill_value_casted = col.dtype.type(fill_value)
            if not np.isnan(fill_value) and (fill_value_casted != fill_value):
>               raise TypeError(
                    f"Cannot safely cast non-equivalent "
                    f"{type(fill_value).__name__} to {col.dtype.name}"
                )
E               TypeError: Cannot safely cast non-equivalent float to int64

/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py:487: TypeError
------------------------------ Captured log call -------------------------------
ERROR    nvtabular:workflow.py:338 Failed to transform operator <nvtabular.ops.fill.FillMedian object at 0x7fef784c2cd0>
Traceback (most recent call last):
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/workflow.py", line 336, in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
  File "/conda/envs/rapids/lib/python3.8/contextlib.py", line 75, in inner
    return func(*args, **kwds)
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/ops/fill.py", line 98, in transform
    df[col] = df[col].fillna(self.medians[col])
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py", line 1874, in fillna
    return super().fillna(
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py", line 1475, in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py", line 487, in fillna
    raise TypeError(
TypeError: Cannot safely cast non-equivalent float to int64
____________________ test_tf_gpu_dl[False-100-parquet-0.06] ____________________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-55/test_tf_gpu_dl_False_100_parqu1')
paths = ['/tmp/pytest-of-jenkins/pytest-55/parquet0/dataset-0.parquet', '/tmp/pytest-of-jenkins/pytest-55/parquet0/dataset-1.parquet']
use_paths = False
dataset = <nvtabular.io.dataset.Dataset object at 0x7ff02817b1c0>
batch_size = 100, gpu_memory_frac = 0.06, engine = 'parquet'

    @pytest.mark.parametrize("gpu_memory_frac", [0.01, 0.06])
    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("batch_size", [1, 10, 100])
    @pytest.mark.parametrize("use_paths", [True, False])
    def test_tf_gpu_dl(tmpdir, paths, use_paths, dataset, batch_size, gpu_memory_frac, engine):
        cont_names = ["x", "y", "id"]
        cat_names = ["name-string"]
        label_name = ["label"]
        if engine == "parquet":
            cat_names.append("name-cat")
    
        columns = cont_names + cat_names
    
        conts = cont_names >> ops.FillMedian() >> ops.Normalize()
        cats = cat_names >> ops.Categorify()
    
        workflow = nvt.Workflow(conts + cats + label_name)
>       workflow.fit(dataset)

tests/unit/test_tf_dataloader.py:91: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:144: in fit
    results = dask.compute(stats, scheduler="synchronous")[0]
/conda/envs/rapids/lib/python3.8/site-packages/dask/base.py:563: in compute
    results = schedule(dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:528: in get_sync
    return get_async(apply_sync, 1, dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:495: in get_async
    fire_task()
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:457: in fire_task
    apply_async(
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:517: in apply_sync
    res = func(*args, **kwds)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:227: in execute_task
    result = pack_exception(e, dumps)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:222: in execute_task
    result = _execute_task(task, data)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/optimization.py:963: in __call__
    return core.get(self.dsk, self.outkey, dict(zip(self.inkeys, args)))
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:151: in get
    result = _execute_task(task, cache)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/utils.py:35: in apply
    return func(*args, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/dataframe/core.py:5471: in apply_and_enforce
    df = func(*args, **kwargs)
nvtabular/workflow.py:336: in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
nvtabular/ops/fill.py:98: in transform
    df[col] = df[col].fillna(self.medians[col])
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py:1874: in fillna
    return super().fillna(
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py:1475: in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <cudf.core.column.numerical.NumericalColumn object at 0x7ff0402c5840>
fill_value = 1000.5, method = None, dtype = None, fill_nan = True

    def fillna(
        self,
        fill_value: Any = None,
        method: str = None,
        dtype: Dtype = None,
        fill_nan: bool = True,
    ) -> NumericalColumn:
        """
        Fill null values with *fill_value*
        """
        if fill_nan:
            col = self.nans_to_nulls()
        else:
            col = self
    
        if method is not None:
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if fill_value is None:
            raise ValueError("Must specify either 'fill_value' or 'method'")
    
        if (
            isinstance(fill_value, cudf.Scalar)
            and fill_value.dtype == col.dtype
        ):
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if np.isscalar(fill_value):
            # cast safely to the same dtype as self
            fill_value_casted = col.dtype.type(fill_value)
            if not np.isnan(fill_value) and (fill_value_casted != fill_value):
>               raise TypeError(
                    f"Cannot safely cast non-equivalent "
                    f"{type(fill_value).__name__} to {col.dtype.name}"
                )
E               TypeError: Cannot safely cast non-equivalent float to int64

/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py:487: TypeError
------------------------------ Captured log call -------------------------------
ERROR    nvtabular:workflow.py:338 Failed to transform operator <nvtabular.ops.fill.FillMedian object at 0x7ff02817b6a0>
Traceback (most recent call last):
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/workflow.py", line 336, in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
  File "/conda/envs/rapids/lib/python3.8/contextlib.py", line 75, in inner
    return func(*args, **kwds)
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/ops/fill.py", line 98, in transform
    df[col] = df[col].fillna(self.medians[col])
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py", line 1874, in fillna
    return super().fillna(
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py", line 1475, in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py", line 487, in fillna
    raise TypeError(
TypeError: Cannot safely cast non-equivalent float to int64
_________ test_empty_cols[label_name0-cont_names0-cat_names0-parquet] __________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-55/test_empty_cols_label_name0_co0')
df =       name-cat name-string    id  label         x         y
0        Quinn        Gary  1020    996  0.672772 -0.59399...t  1027   1048  0.668489 -0.437601
4320    Xavier     Michael   996   1040  0.471682  0.617638

[4321 rows x 6 columns]
dataset = <nvtabular.io.dataset.Dataset object at 0x7ff028287c40>
engine = 'parquet', cat_names = ['name-cat', 'name-string']
cont_names = ['x', 'y', 'id'], label_name = ['label']

    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("cat_names", [["name-cat", "name-string"], []])
    @pytest.mark.parametrize("cont_names", [["x", "y", "id"], []])
    @pytest.mark.parametrize("label_name", [["label"], []])
    def test_empty_cols(tmpdir, df, dataset, engine, cat_names, cont_names, label_name):
    
        features = []
        if cont_names:
            features.append(cont_names >> ops.FillMedian() >> ops.Normalize())
        if cat_names:
            features.append(cat_names >> ops.Categorify())
    
        # test out https://github.com/NVIDIA/NVTabular/issues/149 making sure we can iterate over
        # empty cats/conts
        graph = sum(features, nvt.ColumnGroup(label_name))
        if not graph.columns:
            # if we don't have conts/cats/labels we're done
            return
    
        processor = nvt.Workflow(sum(features, nvt.ColumnGroup(label_name)))
    
        output_train = os.path.join(tmpdir, "train/")
        os.mkdir(output_train)
    
>       df_out = processor.fit_transform(dataset).to_ddf().compute(scheduler="synchronous")

tests/unit/test_torch_dataloader.py:76: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:177: in fit_transform
    self.fit(dataset)
nvtabular/workflow.py:144: in fit
    results = dask.compute(stats, scheduler="synchronous")[0]
/conda/envs/rapids/lib/python3.8/site-packages/dask/base.py:563: in compute
    results = schedule(dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:528: in get_sync
    return get_async(apply_sync, 1, dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:495: in get_async
    fire_task()
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:457: in fire_task
    apply_async(
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:517: in apply_sync
    res = func(*args, **kwds)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:227: in execute_task
    result = pack_exception(e, dumps)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:222: in execute_task
    result = _execute_task(task, data)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/optimization.py:963: in __call__
    return core.get(self.dsk, self.outkey, dict(zip(self.inkeys, args)))
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:151: in get
    result = _execute_task(task, cache)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/utils.py:35: in apply
    return func(*args, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/dataframe/core.py:5471: in apply_and_enforce
    df = func(*args, **kwargs)
nvtabular/workflow.py:336: in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
nvtabular/ops/fill.py:98: in transform
    df[col] = df[col].fillna(self.medians[col])
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py:1874: in fillna
    return super().fillna(
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py:1475: in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <cudf.core.column.numerical.NumericalColumn object at 0x7ff0280a5b40>
fill_value = 1000.5, method = None, dtype = None, fill_nan = True

    def fillna(
        self,
        fill_value: Any = None,
        method: str = None,
        dtype: Dtype = None,
        fill_nan: bool = True,
    ) -> NumericalColumn:
        """
        Fill null values with *fill_value*
        """
        if fill_nan:
            col = self.nans_to_nulls()
        else:
            col = self
    
        if method is not None:
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if fill_value is None:
            raise ValueError("Must specify either 'fill_value' or 'method'")
    
        if (
            isinstance(fill_value, cudf.Scalar)
            and fill_value.dtype == col.dtype
        ):
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if np.isscalar(fill_value):
            # cast safely to the same dtype as self
            fill_value_casted = col.dtype.type(fill_value)
            if not np.isnan(fill_value) and (fill_value_casted != fill_value):
>               raise TypeError(
                    f"Cannot safely cast non-equivalent "
                    f"{type(fill_value).__name__} to {col.dtype.name}"
                )
E               TypeError: Cannot safely cast non-equivalent float to int64

/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py:487: TypeError
------------------------------ Captured log call -------------------------------
ERROR    nvtabular:workflow.py:338 Failed to transform operator <nvtabular.ops.fill.FillMedian object at 0x7ff0282876a0>
Traceback (most recent call last):
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/workflow.py", line 336, in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
  File "/conda/envs/rapids/lib/python3.8/contextlib.py", line 75, in inner
    return func(*args, **kwds)
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/ops/fill.py", line 98, in transform
    df[col] = df[col].fillna(self.medians[col])
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py", line 1874, in fillna
    return super().fillna(
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py", line 1475, in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py", line 487, in fillna
    raise TypeError(
TypeError: Cannot safely cast non-equivalent float to int64
_________ test_empty_cols[label_name0-cont_names0-cat_names1-parquet] __________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-55/test_empty_cols_label_name0_co1')
df =       name-cat name-string    id  label         x         y
0        Quinn        Gary  1020    996  0.672772 -0.59399...t  1027   1048  0.668489 -0.437601
4320    Xavier     Michael   996   1040  0.471682  0.617638

[4321 rows x 6 columns]
dataset = <nvtabular.io.dataset.Dataset object at 0x7fef58429520>
engine = 'parquet', cat_names = [], cont_names = ['x', 'y', 'id']
label_name = ['label']

    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("cat_names", [["name-cat", "name-string"], []])
    @pytest.mark.parametrize("cont_names", [["x", "y", "id"], []])
    @pytest.mark.parametrize("label_name", [["label"], []])
    def test_empty_cols(tmpdir, df, dataset, engine, cat_names, cont_names, label_name):
    
        features = []
        if cont_names:
            features.append(cont_names >> ops.FillMedian() >> ops.Normalize())
        if cat_names:
            features.append(cat_names >> ops.Categorify())
    
        # test out https://github.com/NVIDIA/NVTabular/issues/149 making sure we can iterate over
        # empty cats/conts
        graph = sum(features, nvt.ColumnGroup(label_name))
        if not graph.columns:
            # if we don't have conts/cats/labels we're done
            return
    
        processor = nvt.Workflow(sum(features, nvt.ColumnGroup(label_name)))
    
        output_train = os.path.join(tmpdir, "train/")
        os.mkdir(output_train)
    
>       df_out = processor.fit_transform(dataset).to_ddf().compute(scheduler="synchronous")

tests/unit/test_torch_dataloader.py:76: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:177: in fit_transform
    self.fit(dataset)
nvtabular/workflow.py:147: in fit
    op.fit_finalize(computed_stats)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <nvtabular.ops.fill.FillMedian object at 0x7fef584295b0>
dask_stats = x     0.017761632
y            <NA>
id         1000.5
Name: 0.5, dtype: float64

    @annotate("FillMedian_finalize", color="green", domain="nvt_python")
    def fit_finalize(self, dask_stats):
        index = dask_stats.index
        vals = index.values_host if hasattr(index, "values_host") else index.values
        for col in vals:
>           self.medians[col] = float(dask_stats[col])
E           TypeError: float() argument must be a string or a number, not '_NAType'

nvtabular/ops/fill.py:112: TypeError
_________ test_empty_cols[label_name1-cont_names0-cat_names0-parquet] __________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-55/test_empty_cols_label_name1_co0')
df =       name-cat name-string    id  label         x         y
0        Quinn        Gary  1020    996  0.672772 -0.59399...t  1027   1048  0.668489 -0.437601
4320    Xavier     Michael   996   1040  0.471682  0.617638

[4321 rows x 6 columns]
dataset = <nvtabular.io.dataset.Dataset object at 0x7ff018292610>
engine = 'parquet', cat_names = ['name-cat', 'name-string']
cont_names = ['x', 'y', 'id'], label_name = []

    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("cat_names", [["name-cat", "name-string"], []])
    @pytest.mark.parametrize("cont_names", [["x", "y", "id"], []])
    @pytest.mark.parametrize("label_name", [["label"], []])
    def test_empty_cols(tmpdir, df, dataset, engine, cat_names, cont_names, label_name):
    
        features = []
        if cont_names:
            features.append(cont_names >> ops.FillMedian() >> ops.Normalize())
        if cat_names:
            features.append(cat_names >> ops.Categorify())
    
        # test out https://github.com/NVIDIA/NVTabular/issues/149 making sure we can iterate over
        # empty cats/conts
        graph = sum(features, nvt.ColumnGroup(label_name))
        if not graph.columns:
            # if we don't have conts/cats/labels we're done
            return
    
        processor = nvt.Workflow(sum(features, nvt.ColumnGroup(label_name)))
    
        output_train = os.path.join(tmpdir, "train/")
        os.mkdir(output_train)
    
>       df_out = processor.fit_transform(dataset).to_ddf().compute(scheduler="synchronous")

tests/unit/test_torch_dataloader.py:76: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:177: in fit_transform
    self.fit(dataset)
nvtabular/workflow.py:144: in fit
    results = dask.compute(stats, scheduler="synchronous")[0]
/conda/envs/rapids/lib/python3.8/site-packages/dask/base.py:563: in compute
    results = schedule(dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:528: in get_sync
    return get_async(apply_sync, 1, dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:495: in get_async
    fire_task()
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:457: in fire_task
    apply_async(
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:517: in apply_sync
    res = func(*args, **kwds)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:227: in execute_task
    result = pack_exception(e, dumps)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:222: in execute_task
    result = _execute_task(task, data)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/optimization.py:963: in __call__
    return core.get(self.dsk, self.outkey, dict(zip(self.inkeys, args)))
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:151: in get
    result = _execute_task(task, cache)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/utils.py:35: in apply
    return func(*args, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/dataframe/core.py:5471: in apply_and_enforce
    df = func(*args, **kwargs)
nvtabular/workflow.py:336: in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
nvtabular/ops/fill.py:98: in transform
    df[col] = df[col].fillna(self.medians[col])
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py:1874: in fillna
    return super().fillna(
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py:1475: in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <cudf.core.column.numerical.NumericalColumn object at 0x7ff02829f840>
fill_value = 1000.5, method = None, dtype = None, fill_nan = True

    def fillna(
        self,
        fill_value: Any = None,
        method: str = None,
        dtype: Dtype = None,
        fill_nan: bool = True,
    ) -> NumericalColumn:
        """
        Fill null values with *fill_value*
        """
        if fill_nan:
            col = self.nans_to_nulls()
        else:
            col = self
    
        if method is not None:
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if fill_value is None:
            raise ValueError("Must specify either 'fill_value' or 'method'")
    
        if (
            isinstance(fill_value, cudf.Scalar)
            and fill_value.dtype == col.dtype
        ):
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if np.isscalar(fill_value):
            # cast safely to the same dtype as self
            fill_value_casted = col.dtype.type(fill_value)
            if not np.isnan(fill_value) and (fill_value_casted != fill_value):
>               raise TypeError(
                    f"Cannot safely cast non-equivalent "
                    f"{type(fill_value).__name__} to {col.dtype.name}"
                )
E               TypeError: Cannot safely cast non-equivalent float to int64

/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py:487: TypeError
------------------------------ Captured log call -------------------------------
ERROR    nvtabular:workflow.py:338 Failed to transform operator <nvtabular.ops.fill.FillMedian object at 0x7fef782d4e50>
Traceback (most recent call last):
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/workflow.py", line 336, in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
  File "/conda/envs/rapids/lib/python3.8/contextlib.py", line 75, in inner
    return func(*args, **kwds)
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/ops/fill.py", line 98, in transform
    df[col] = df[col].fillna(self.medians[col])
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py", line 1874, in fillna
    return super().fillna(
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py", line 1475, in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py", line 487, in fillna
    raise TypeError(
TypeError: Cannot safely cast non-equivalent float to int64
_________ test_empty_cols[label_name1-cont_names0-cat_names1-parquet] __________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-55/test_empty_cols_label_name1_co1')
df =       name-cat name-string    id  label         x         y
0        Quinn        Gary  1020    996  0.672772 -0.59399...t  1027   1048  0.668489 -0.437601
4320    Xavier     Michael   996   1040  0.471682  0.617638

[4321 rows x 6 columns]
dataset = <nvtabular.io.dataset.Dataset object at 0x7ff105c96460>
engine = 'parquet', cat_names = [], cont_names = ['x', 'y', 'id']
label_name = []

    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("cat_names", [["name-cat", "name-string"], []])
    @pytest.mark.parametrize("cont_names", [["x", "y", "id"], []])
    @pytest.mark.parametrize("label_name", [["label"], []])
    def test_empty_cols(tmpdir, df, dataset, engine, cat_names, cont_names, label_name):
    
        features = []
        if cont_names:
            features.append(cont_names >> ops.FillMedian() >> ops.Normalize())
        if cat_names:
            features.append(cat_names >> ops.Categorify())
    
        # test out https://github.com/NVIDIA/NVTabular/issues/149 making sure we can iterate over
        # empty cats/conts
        graph = sum(features, nvt.ColumnGroup(label_name))
        if not graph.columns:
            # if we don't have conts/cats/labels we're done
            return
    
        processor = nvt.Workflow(sum(features, nvt.ColumnGroup(label_name)))
    
        output_train = os.path.join(tmpdir, "train/")
        os.mkdir(output_train)
    
>       df_out = processor.fit_transform(dataset).to_ddf().compute(scheduler="synchronous")

tests/unit/test_torch_dataloader.py:76: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:177: in fit_transform
    self.fit(dataset)
nvtabular/workflow.py:144: in fit
    results = dask.compute(stats, scheduler="synchronous")[0]
/conda/envs/rapids/lib/python3.8/site-packages/dask/base.py:563: in compute
    results = schedule(dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:528: in get_sync
    return get_async(apply_sync, 1, dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:495: in get_async
    fire_task()
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:457: in fire_task
    apply_async(
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:517: in apply_sync
    res = func(*args, **kwds)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:227: in execute_task
    result = pack_exception(e, dumps)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:222: in execute_task
    result = _execute_task(task, data)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/optimization.py:963: in __call__
    return core.get(self.dsk, self.outkey, dict(zip(self.inkeys, args)))
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:151: in get
    result = _execute_task(task, cache)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/utils.py:35: in apply
    return func(*args, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/dataframe/core.py:5471: in apply_and_enforce
    df = func(*args, **kwargs)
nvtabular/workflow.py:336: in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
nvtabular/ops/fill.py:98: in transform
    df[col] = df[col].fillna(self.medians[col])
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py:1874: in fillna
    return super().fillna(
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py:1475: in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <cudf.core.column.numerical.NumericalColumn object at 0x7ff028081840>
fill_value = 1000.5, method = None, dtype = None, fill_nan = True

    def fillna(
        self,
        fill_value: Any = None,
        method: str = None,
        dtype: Dtype = None,
        fill_nan: bool = True,
    ) -> NumericalColumn:
        """
        Fill null values with *fill_value*
        """
        if fill_nan:
            col = self.nans_to_nulls()
        else:
            col = self
    
        if method is not None:
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if fill_value is None:
            raise ValueError("Must specify either 'fill_value' or 'method'")
    
        if (
            isinstance(fill_value, cudf.Scalar)
            and fill_value.dtype == col.dtype
        ):
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if np.isscalar(fill_value):
            # cast safely to the same dtype as self
            fill_value_casted = col.dtype.type(fill_value)
            if not np.isnan(fill_value) and (fill_value_casted != fill_value):
>               raise TypeError(
                    f"Cannot safely cast non-equivalent "
                    f"{type(fill_value).__name__} to {col.dtype.name}"
                )
E               TypeError: Cannot safely cast non-equivalent float to int64

/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py:487: TypeError
------------------------------ Captured log call -------------------------------
ERROR    nvtabular:workflow.py:338 Failed to transform operator <nvtabular.ops.fill.FillMedian object at 0x7ff105c964f0>
Traceback (most recent call last):
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/workflow.py", line 336, in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
  File "/conda/envs/rapids/lib/python3.8/contextlib.py", line 75, in inner
    return func(*args, **kwds)
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/ops/fill.py", line 98, in transform
    df[col] = df[col].fillna(self.medians[col])
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py", line 1874, in fillna
    return super().fillna(
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py", line 1475, in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py", line 487, in fillna
    raise TypeError(
TypeError: Cannot safely cast non-equivalent float to int64
__________________ test_gpu_dl_break[None-parquet-1000-0.001] __________________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-55/test_gpu_dl_break_None_parquet0')
df =       name-cat name-string    id  label         x         y
0        Quinn        Gary  1020    996  0.672772 -0.59399...t  1027   1048  0.668489 -0.437601
4320    Xavier     Michael   996   1040  0.471682  0.617638

[4321 rows x 6 columns]
dataset = <nvtabular.io.dataset.Dataset object at 0x7ff105d580d0>
batch_size = 1000, part_mem_fraction = 0.001, engine = 'parquet', device = None

    @pytest.mark.parametrize("part_mem_fraction", [0.001, 0.06])
    @pytest.mark.parametrize("batch_size", [1000])
    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("device", [None, 0])
    def test_gpu_dl_break(tmpdir, df, dataset, batch_size, part_mem_fraction, engine, device):
        cat_names = ["name-cat", "name-string"]
        cont_names = ["x", "y", "id"]
        label_name = ["label"]
    
        conts = cont_names >> ops.FillMedian() >> ops.Normalize()
        cats = cat_names >> ops.Categorify()
    
        processor = nvt.Workflow(conts + cats + label_name)
    
        output_train = os.path.join(tmpdir, "train/")
        os.mkdir(output_train)
    
>       processor.fit_transform(dataset).to_parquet(
            shuffle=nvt.io.Shuffle.PER_PARTITION,
            output_path=output_train,
            out_files_per_proc=2,
        )

tests/unit/test_torch_dataloader.py:109: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:177: in fit_transform
    self.fit(dataset)
nvtabular/workflow.py:144: in fit
    results = dask.compute(stats, scheduler="synchronous")[0]
/conda/envs/rapids/lib/python3.8/site-packages/dask/base.py:563: in compute
    results = schedule(dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:528: in get_sync
    return get_async(apply_sync, 1, dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:495: in get_async
    fire_task()
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:457: in fire_task
    apply_async(
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:517: in apply_sync
    res = func(*args, **kwds)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:227: in execute_task
    result = pack_exception(e, dumps)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:222: in execute_task
    result = _execute_task(task, data)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/optimization.py:963: in __call__
    return core.get(self.dsk, self.outkey, dict(zip(self.inkeys, args)))
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:151: in get
    result = _execute_task(task, cache)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/utils.py:35: in apply
    return func(*args, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/dataframe/core.py:5471: in apply_and_enforce
    df = func(*args, **kwargs)
nvtabular/workflow.py:336: in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
nvtabular/ops/fill.py:98: in transform
    df[col] = df[col].fillna(self.medians[col])
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py:1874: in fillna
    return super().fillna(
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py:1475: in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <cudf.core.column.numerical.NumericalColumn object at 0x7ff040281d40>
fill_value = 1000.5, method = None, dtype = None, fill_nan = True

    def fillna(
        self,
        fill_value: Any = None,
        method: str = None,
        dtype: Dtype = None,
        fill_nan: bool = True,
    ) -> NumericalColumn:
        """
        Fill null values with *fill_value*
        """
        if fill_nan:
            col = self.nans_to_nulls()
        else:
            col = self
    
        if method is not None:
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if fill_value is None:
            raise ValueError("Must specify either 'fill_value' or 'method'")
    
        if (
            isinstance(fill_value, cudf.Scalar)
            and fill_value.dtype == col.dtype
        ):
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if np.isscalar(fill_value):
            # cast safely to the same dtype as self
            fill_value_casted = col.dtype.type(fill_value)
            if not np.isnan(fill_value) and (fill_value_casted != fill_value):
>               raise TypeError(
                    f"Cannot safely cast non-equivalent "
                    f"{type(fill_value).__name__} to {col.dtype.name}"
                )
E               TypeError: Cannot safely cast non-equivalent float to int64

/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py:487: TypeError
------------------------------ Captured log call -------------------------------
ERROR    nvtabular:workflow.py:338 Failed to transform operator <nvtabular.ops.fill.FillMedian object at 0x7ff105d58f70>
Traceback (most recent call last):
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/workflow.py", line 336, in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
  File "/conda/envs/rapids/lib/python3.8/contextlib.py", line 75, in inner
    return func(*args, **kwds)
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/ops/fill.py", line 98, in transform
    df[col] = df[col].fillna(self.medians[col])
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py", line 1874, in fillna
    return super().fillna(
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py", line 1475, in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py", line 487, in fillna
    raise TypeError(
TypeError: Cannot safely cast non-equivalent float to int64
__________________ test_gpu_dl_break[None-parquet-1000-0.06] ___________________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-55/test_gpu_dl_break_None_parquet1')
df =       name-cat name-string    id  label         x         y
0        Quinn        Gary  1020    996  0.672772 -0.59399...t  1027   1048  0.668489 -0.437601
4320    Xavier     Michael   996   1040  0.471682  0.617638

[4321 rows x 6 columns]
dataset = <nvtabular.io.dataset.Dataset object at 0x7ff105c6d7f0>
batch_size = 1000, part_mem_fraction = 0.06, engine = 'parquet', device = None

    @pytest.mark.parametrize("part_mem_fraction", [0.001, 0.06])
    @pytest.mark.parametrize("batch_size", [1000])
    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("device", [None, 0])
    def test_gpu_dl_break(tmpdir, df, dataset, batch_size, part_mem_fraction, engine, device):
        cat_names = ["name-cat", "name-string"]
        cont_names = ["x", "y", "id"]
        label_name = ["label"]
    
        conts = cont_names >> ops.FillMedian() >> ops.Normalize()
        cats = cat_names >> ops.Categorify()
    
        processor = nvt.Workflow(conts + cats + label_name)
    
        output_train = os.path.join(tmpdir, "train/")
        os.mkdir(output_train)
    
>       processor.fit_transform(dataset).to_parquet(
            shuffle=nvt.io.Shuffle.PER_PARTITION,
            output_path=output_train,
            out_files_per_proc=2,
        )

tests/unit/test_torch_dataloader.py:109: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:177: in fit_transform
    self.fit(dataset)
nvtabular/workflow.py:144: in fit
    results = dask.compute(stats, scheduler="synchronous")[0]
/conda/envs/rapids/lib/python3.8/site-packages/dask/base.py:563: in compute
    results = schedule(dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:528: in get_sync
    return get_async(apply_sync, 1, dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:495: in get_async
    fire_task()
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:457: in fire_task
    apply_async(
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:517: in apply_sync
    res = func(*args, **kwds)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:227: in execute_task
    result = pack_exception(e, dumps)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:222: in execute_task
    result = _execute_task(task, data)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/optimization.py:963: in __call__
    return core.get(self.dsk, self.outkey, dict(zip(self.inkeys, args)))
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:151: in get
    result = _execute_task(task, cache)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/utils.py:35: in apply
    return func(*args, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/dataframe/core.py:5471: in apply_and_enforce
    df = func(*args, **kwargs)
nvtabular/workflow.py:336: in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
nvtabular/ops/fill.py:98: in transform
    df[col] = df[col].fillna(self.medians[col])
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py:1874: in fillna
    return super().fillna(
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py:1475: in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <cudf.core.column.numerical.NumericalColumn object at 0x7ff038a97440>
fill_value = 1000.5, method = None, dtype = None, fill_nan = True

    def fillna(
        self,
        fill_value: Any = None,
        method: str = None,
        dtype: Dtype = None,
        fill_nan: bool = True,
    ) -> NumericalColumn:
        """
        Fill null values with *fill_value*
        """
        if fill_nan:
            col = self.nans_to_nulls()
        else:
            col = self
    
        if method is not None:
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if fill_value is None:
            raise ValueError("Must specify either 'fill_value' or 'method'")
    
        if (
            isinstance(fill_value, cudf.Scalar)
            and fill_value.dtype == col.dtype
        ):
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if np.isscalar(fill_value):
            # cast safely to the same dtype as self
            fill_value_casted = col.dtype.type(fill_value)
            if not np.isnan(fill_value) and (fill_value_casted != fill_value):
>               raise TypeError(
                    f"Cannot safely cast non-equivalent "
                    f"{type(fill_value).__name__} to {col.dtype.name}"
                )
E               TypeError: Cannot safely cast non-equivalent float to int64

/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py:487: TypeError
------------------------------ Captured log call -------------------------------
ERROR    nvtabular:workflow.py:338 Failed to transform operator <nvtabular.ops.fill.FillMedian object at 0x7ff105c6d5b0>
Traceback (most recent call last):
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/workflow.py", line 336, in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
  File "/conda/envs/rapids/lib/python3.8/contextlib.py", line 75, in inner
    return func(*args, **kwds)
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/ops/fill.py", line 98, in transform
    df[col] = df[col].fillna(self.medians[col])
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py", line 1874, in fillna
    return super().fillna(
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py", line 1475, in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py", line 487, in fillna
    raise TypeError(
TypeError: Cannot safely cast non-equivalent float to int64
___________________ test_gpu_dl_break[0-parquet-1000-0.001] ____________________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-55/test_gpu_dl_break_0_parquet_100')
df =       name-cat name-string    id  label         x         y
0        Quinn        Gary  1020    996  0.672772 -0.59399...t  1027   1048  0.668489 -0.437601
4320    Xavier     Michael   996   1040  0.471682  0.617638

[4321 rows x 6 columns]
dataset = <nvtabular.io.dataset.Dataset object at 0x7ff105f296a0>
batch_size = 1000, part_mem_fraction = 0.001, engine = 'parquet', device = 0

    @pytest.mark.parametrize("part_mem_fraction", [0.001, 0.06])
    @pytest.mark.parametrize("batch_size", [1000])
    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("device", [None, 0])
    def test_gpu_dl_break(tmpdir, df, dataset, batch_size, part_mem_fraction, engine, device):
        cat_names = ["name-cat", "name-string"]
        cont_names = ["x", "y", "id"]
        label_name = ["label"]
    
        conts = cont_names >> ops.FillMedian() >> ops.Normalize()
        cats = cat_names >> ops.Categorify()
    
        processor = nvt.Workflow(conts + cats + label_name)
    
        output_train = os.path.join(tmpdir, "train/")
        os.mkdir(output_train)
    
>       processor.fit_transform(dataset).to_parquet(
            shuffle=nvt.io.Shuffle.PER_PARTITION,
            output_path=output_train,
            out_files_per_proc=2,
        )

tests/unit/test_torch_dataloader.py:109: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:177: in fit_transform
    self.fit(dataset)
nvtabular/workflow.py:144: in fit
    results = dask.compute(stats, scheduler="synchronous")[0]
/conda/envs/rapids/lib/python3.8/site-packages/dask/base.py:563: in compute
    results = schedule(dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:528: in get_sync
    return get_async(apply_sync, 1, dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:495: in get_async
    fire_task()
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:457: in fire_task
    apply_async(
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:517: in apply_sync
    res = func(*args, **kwds)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:227: in execute_task
    result = pack_exception(e, dumps)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:222: in execute_task
    result = _execute_task(task, data)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/optimization.py:963: in __call__
    return core.get(self.dsk, self.outkey, dict(zip(self.inkeys, args)))
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:151: in get
    result = _execute_task(task, cache)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/utils.py:35: in apply
    return func(*args, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/dataframe/core.py:5471: in apply_and_enforce
    df = func(*args, **kwargs)
nvtabular/workflow.py:336: in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
nvtabular/ops/fill.py:98: in transform
    df[col] = df[col].fillna(self.medians[col])
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py:1874: in fillna
    return super().fillna(
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py:1475: in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <cudf.core.column.numerical.NumericalColumn object at 0x7ff033a66a40>
fill_value = 1000.5, method = None, dtype = None, fill_nan = True

    def fillna(
        self,
        fill_value: Any = None,
        method: str = None,
        dtype: Dtype = None,
        fill_nan: bool = True,
    ) -> NumericalColumn:
        """
        Fill null values with *fill_value*
        """
        if fill_nan:
            col = self.nans_to_nulls()
        else:
            col = self
    
        if method is not None:
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if fill_value is None:
            raise ValueError("Must specify either 'fill_value' or 'method'")
    
        if (
            isinstance(fill_value, cudf.Scalar)
            and fill_value.dtype == col.dtype
        ):
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if np.isscalar(fill_value):
            # cast safely to the same dtype as self
            fill_value_casted = col.dtype.type(fill_value)
            if not np.isnan(fill_value) and (fill_value_casted != fill_value):
>               raise TypeError(
                    f"Cannot safely cast non-equivalent "
                    f"{type(fill_value).__name__} to {col.dtype.name}"
                )
E               TypeError: Cannot safely cast non-equivalent float to int64

/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py:487: TypeError
------------------------------ Captured log call -------------------------------
ERROR    nvtabular:workflow.py:338 Failed to transform operator <nvtabular.ops.fill.FillMedian object at 0x7ff105f29490>
Traceback (most recent call last):
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/workflow.py", line 336, in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
  File "/conda/envs/rapids/lib/python3.8/contextlib.py", line 75, in inner
    return func(*args, **kwds)
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/ops/fill.py", line 98, in transform
    df[col] = df[col].fillna(self.medians[col])
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py", line 1874, in fillna
    return super().fillna(
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py", line 1475, in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py", line 487, in fillna
    raise TypeError(
TypeError: Cannot safely cast non-equivalent float to int64
____________________ test_gpu_dl_break[0-parquet-1000-0.06] ____________________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-55/test_gpu_dl_break_0_parquet_101')
df =       name-cat name-string    id  label         x         y
0        Quinn        Gary  1020    996  0.672772 -0.59399...t  1027   1048  0.668489 -0.437601
4320    Xavier     Michael   996   1040  0.471682  0.617638

[4321 rows x 6 columns]
dataset = <nvtabular.io.dataset.Dataset object at 0x7ff018782790>
batch_size = 1000, part_mem_fraction = 0.06, engine = 'parquet', device = 0

    @pytest.mark.parametrize("part_mem_fraction", [0.001, 0.06])
    @pytest.mark.parametrize("batch_size", [1000])
    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("device", [None, 0])
    def test_gpu_dl_break(tmpdir, df, dataset, batch_size, part_mem_fraction, engine, device):
        cat_names = ["name-cat", "name-string"]
        cont_names = ["x", "y", "id"]
        label_name = ["label"]
    
        conts = cont_names >> ops.FillMedian() >> ops.Normalize()
        cats = cat_names >> ops.Categorify()
    
        processor = nvt.Workflow(conts + cats + label_name)
    
        output_train = os.path.join(tmpdir, "train/")
        os.mkdir(output_train)
    
>       processor.fit_transform(dataset).to_parquet(
            shuffle=nvt.io.Shuffle.PER_PARTITION,
            output_path=output_train,
            out_files_per_proc=2,
        )

tests/unit/test_torch_dataloader.py:109: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:177: in fit_transform
    self.fit(dataset)
nvtabular/workflow.py:144: in fit
    results = dask.compute(stats, scheduler="synchronous")[0]
/conda/envs/rapids/lib/python3.8/site-packages/dask/base.py:563: in compute
    results = schedule(dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:528: in get_sync
    return get_async(apply_sync, 1, dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:495: in get_async
    fire_task()
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:457: in fire_task
    apply_async(
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:517: in apply_sync
    res = func(*args, **kwds)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:227: in execute_task
    result = pack_exception(e, dumps)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:222: in execute_task
    result = _execute_task(task, data)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/optimization.py:963: in __call__
    return core.get(self.dsk, self.outkey, dict(zip(self.inkeys, args)))
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:151: in get
    result = _execute_task(task, cache)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/utils.py:35: in apply
    return func(*args, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/dataframe/core.py:5471: in apply_and_enforce
    df = func(*args, **kwargs)
nvtabular/workflow.py:336: in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
nvtabular/ops/fill.py:98: in transform
    df[col] = df[col].fillna(self.medians[col])
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py:1874: in fillna
    return super().fillna(
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py:1475: in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <cudf.core.column.numerical.NumericalColumn object at 0x7ff038228240>
fill_value = 1000.5, method = None, dtype = None, fill_nan = True

    def fillna(
        self,
        fill_value: Any = None,
        method: str = None,
        dtype: Dtype = None,
        fill_nan: bool = True,
    ) -> NumericalColumn:
        """
        Fill null values with *fill_value*
        """
        if fill_nan:
            col = self.nans_to_nulls()
        else:
            col = self
    
        if method is not None:
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if fill_value is None:
            raise ValueError("Must specify either 'fill_value' or 'method'")
    
        if (
            isinstance(fill_value, cudf.Scalar)
            and fill_value.dtype == col.dtype
        ):
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if np.isscalar(fill_value):
            # cast safely to the same dtype as self
            fill_value_casted = col.dtype.type(fill_value)
            if not np.isnan(fill_value) and (fill_value_casted != fill_value):
>               raise TypeError(
                    f"Cannot safely cast non-equivalent "
                    f"{type(fill_value).__name__} to {col.dtype.name}"
                )
E               TypeError: Cannot safely cast non-equivalent float to int64

/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py:487: TypeError
------------------------------ Captured log call -------------------------------
ERROR    nvtabular:workflow.py:338 Failed to transform operator <nvtabular.ops.fill.FillMedian object at 0x7ff105f6e7f0>
Traceback (most recent call last):
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/workflow.py", line 336, in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
  File "/conda/envs/rapids/lib/python3.8/contextlib.py", line 75, in inner
    return func(*args, **kwds)
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/ops/fill.py", line 98, in transform
    df[col] = df[col].fillna(self.medians[col])
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py", line 1874, in fillna
    return super().fillna(
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py", line 1475, in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py", line 487, in fillna
    raise TypeError(
TypeError: Cannot safely cast non-equivalent float to int64
_____________________ test_gpu_dl[None-parquet-1000-0.001] _____________________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-55/test_gpu_dl_None_parquet_1000_0')
df =       name-cat name-string    id  label         x         y
0        Quinn        Gary  1020    996  0.672772 -0.59399...t  1027   1048  0.668489 -0.437601
4320    Xavier     Michael   996   1040  0.471682  0.617638

[4321 rows x 6 columns]
dataset = <nvtabular.io.dataset.Dataset object at 0x7ff105ed7ac0>
batch_size = 1000, part_mem_fraction = 0.001, engine = 'parquet', device = None

    @pytest.mark.parametrize("part_mem_fraction", [0.001, 0.06])
    @pytest.mark.parametrize("batch_size", [1000])
    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("device", [None, 0])
    def test_gpu_dl(tmpdir, df, dataset, batch_size, part_mem_fraction, engine, device):
        cat_names = ["name-cat", "name-string"]
        cont_names = ["x", "y", "id"]
        label_name = ["label"]
    
        conts = cont_names >> ops.FillMedian() >> ops.Normalize()
        cats = cat_names >> ops.Categorify()
    
        processor = nvt.Workflow(conts + cats + label_name)
    
        output_train = os.path.join(tmpdir, "train/")
        os.mkdir(output_train)
    
>       processor.fit_transform(dataset).to_parquet(
            shuffle=nvt.io.Shuffle.PER_PARTITION,
            output_path=output_train,
            out_files_per_proc=2,
        )

tests/unit/test_torch_dataloader.py:169: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:177: in fit_transform
    self.fit(dataset)
nvtabular/workflow.py:144: in fit
    results = dask.compute(stats, scheduler="synchronous")[0]
/conda/envs/rapids/lib/python3.8/site-packages/dask/base.py:563: in compute
    results = schedule(dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:528: in get_sync
    return get_async(apply_sync, 1, dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:495: in get_async
    fire_task()
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:457: in fire_task
    apply_async(
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:517: in apply_sync
    res = func(*args, **kwds)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:227: in execute_task
    result = pack_exception(e, dumps)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:222: in execute_task
    result = _execute_task(task, data)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/optimization.py:963: in __call__
    return core.get(self.dsk, self.outkey, dict(zip(self.inkeys, args)))
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:151: in get
    result = _execute_task(task, cache)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/utils.py:35: in apply
    return func(*args, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/dataframe/core.py:5471: in apply_and_enforce
    df = func(*args, **kwargs)
nvtabular/workflow.py:336: in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
nvtabular/ops/fill.py:98: in transform
    df[col] = df[col].fillna(self.medians[col])
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py:1874: in fillna
    return super().fillna(
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py:1475: in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <cudf.core.column.numerical.NumericalColumn object at 0x7ff040284f40>
fill_value = 1000.5, method = None, dtype = None, fill_nan = True

    def fillna(
        self,
        fill_value: Any = None,
        method: str = None,
        dtype: Dtype = None,
        fill_nan: bool = True,
    ) -> NumericalColumn:
        """
        Fill null values with *fill_value*
        """
        if fill_nan:
            col = self.nans_to_nulls()
        else:
            col = self
    
        if method is not None:
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if fill_value is None:
            raise ValueError("Must specify either 'fill_value' or 'method'")
    
        if (
            isinstance(fill_value, cudf.Scalar)
            and fill_value.dtype == col.dtype
        ):
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if np.isscalar(fill_value):
            # cast safely to the same dtype as self
            fill_value_casted = col.dtype.type(fill_value)
            if not np.isnan(fill_value) and (fill_value_casted != fill_value):
>               raise TypeError(
                    f"Cannot safely cast non-equivalent "
                    f"{type(fill_value).__name__} to {col.dtype.name}"
                )
E               TypeError: Cannot safely cast non-equivalent float to int64

/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py:487: TypeError
------------------------------ Captured log call -------------------------------
ERROR    nvtabular:workflow.py:338 Failed to transform operator <nvtabular.ops.fill.FillMedian object at 0x7ff105ed7940>
Traceback (most recent call last):
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/workflow.py", line 336, in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
  File "/conda/envs/rapids/lib/python3.8/contextlib.py", line 75, in inner
    return func(*args, **kwds)
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/ops/fill.py", line 98, in transform
    df[col] = df[col].fillna(self.medians[col])
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py", line 1874, in fillna
    return super().fillna(
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py", line 1475, in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py", line 487, in fillna
    raise TypeError(
TypeError: Cannot safely cast non-equivalent float to int64
_____________________ test_gpu_dl[None-parquet-1000-0.06] ______________________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-55/test_gpu_dl_None_parquet_1000_1')
df =       name-cat name-string    id  label         x         y
0        Quinn        Gary  1020    996  0.672772 -0.59399...t  1027   1048  0.668489 -0.437601
4320    Xavier     Michael   996   1040  0.471682  0.617638

[4321 rows x 6 columns]
dataset = <nvtabular.io.dataset.Dataset object at 0x7fef5840f4f0>
batch_size = 1000, part_mem_fraction = 0.06, engine = 'parquet', device = None

    @pytest.mark.parametrize("part_mem_fraction", [0.001, 0.06])
    @pytest.mark.parametrize("batch_size", [1000])
    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("device", [None, 0])
    def test_gpu_dl(tmpdir, df, dataset, batch_size, part_mem_fraction, engine, device):
        cat_names = ["name-cat", "name-string"]
        cont_names = ["x", "y", "id"]
        label_name = ["label"]
    
        conts = cont_names >> ops.FillMedian() >> ops.Normalize()
        cats = cat_names >> ops.Categorify()
    
        processor = nvt.Workflow(conts + cats + label_name)
    
        output_train = os.path.join(tmpdir, "train/")
        os.mkdir(output_train)
    
>       processor.fit_transform(dataset).to_parquet(
            shuffle=nvt.io.Shuffle.PER_PARTITION,
            output_path=output_train,
            out_files_per_proc=2,
        )

tests/unit/test_torch_dataloader.py:169: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:177: in fit_transform
    self.fit(dataset)
nvtabular/workflow.py:144: in fit
    results = dask.compute(stats, scheduler="synchronous")[0]
/conda/envs/rapids/lib/python3.8/site-packages/dask/base.py:563: in compute
    results = schedule(dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:528: in get_sync
    return get_async(apply_sync, 1, dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:495: in get_async
    fire_task()
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:457: in fire_task
    apply_async(
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:517: in apply_sync
    res = func(*args, **kwds)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:227: in execute_task
    result = pack_exception(e, dumps)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:222: in execute_task
    result = _execute_task(task, data)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/optimization.py:963: in __call__
    return core.get(self.dsk, self.outkey, dict(zip(self.inkeys, args)))
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:151: in get
    result = _execute_task(task, cache)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/utils.py:35: in apply
    return func(*args, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/dataframe/core.py:5471: in apply_and_enforce
    df = func(*args, **kwargs)
nvtabular/workflow.py:336: in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
nvtabular/ops/fill.py:98: in transform
    df[col] = df[col].fillna(self.medians[col])
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py:1874: in fillna
    return super().fillna(
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py:1475: in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <cudf.core.column.numerical.NumericalColumn object at 0x7ff040284ac0>
fill_value = 1000.5, method = None, dtype = None, fill_nan = True

    def fillna(
        self,
        fill_value: Any = None,
        method: str = None,
        dtype: Dtype = None,
        fill_nan: bool = True,
    ) -> NumericalColumn:
        """
        Fill null values with *fill_value*
        """
        if fill_nan:
            col = self.nans_to_nulls()
        else:
            col = self
    
        if method is not None:
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if fill_value is None:
            raise ValueError("Must specify either 'fill_value' or 'method'")
    
        if (
            isinstance(fill_value, cudf.Scalar)
            and fill_value.dtype == col.dtype
        ):
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if np.isscalar(fill_value):
            # cast safely to the same dtype as self
            fill_value_casted = col.dtype.type(fill_value)
            if not np.isnan(fill_value) and (fill_value_casted != fill_value):
>               raise TypeError(
                    f"Cannot safely cast non-equivalent "
                    f"{type(fill_value).__name__} to {col.dtype.name}"
                )
E               TypeError: Cannot safely cast non-equivalent float to int64

/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py:487: TypeError
------------------------------ Captured log call -------------------------------
ERROR    nvtabular:workflow.py:338 Failed to transform operator <nvtabular.ops.fill.FillMedian object at 0x7fef5840f2e0>
Traceback (most recent call last):
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/workflow.py", line 336, in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
  File "/conda/envs/rapids/lib/python3.8/contextlib.py", line 75, in inner
    return func(*args, **kwds)
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/ops/fill.py", line 98, in transform
    df[col] = df[col].fillna(self.medians[col])
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py", line 1874, in fillna
    return super().fillna(
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py", line 1475, in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py", line 487, in fillna
    raise TypeError(
TypeError: Cannot safely cast non-equivalent float to int64
______________________ test_gpu_dl[0-parquet-1000-0.001] _______________________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-55/test_gpu_dl_0_parquet_1000_0_00')
df =       name-cat name-string    id  label         x         y
0        Quinn        Gary  1020    996  0.672772 -0.59399...t  1027   1048  0.668489 -0.437601
4320    Xavier     Michael   996   1040  0.471682  0.617638

[4321 rows x 6 columns]
dataset = <nvtabular.io.dataset.Dataset object at 0x7ff1059ff880>
batch_size = 1000, part_mem_fraction = 0.001, engine = 'parquet', device = 0

    @pytest.mark.parametrize("part_mem_fraction", [0.001, 0.06])
    @pytest.mark.parametrize("batch_size", [1000])
    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("device", [None, 0])
    def test_gpu_dl(tmpdir, df, dataset, batch_size, part_mem_fraction, engine, device):
        cat_names = ["name-cat", "name-string"]
        cont_names = ["x", "y", "id"]
        label_name = ["label"]
    
        conts = cont_names >> ops.FillMedian() >> ops.Normalize()
        cats = cat_names >> ops.Categorify()
    
        processor = nvt.Workflow(conts + cats + label_name)
    
        output_train = os.path.join(tmpdir, "train/")
        os.mkdir(output_train)
    
>       processor.fit_transform(dataset).to_parquet(
            shuffle=nvt.io.Shuffle.PER_PARTITION,
            output_path=output_train,
            out_files_per_proc=2,
        )

tests/unit/test_torch_dataloader.py:169: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:177: in fit_transform
    self.fit(dataset)
nvtabular/workflow.py:144: in fit
    results = dask.compute(stats, scheduler="synchronous")[0]
/conda/envs/rapids/lib/python3.8/site-packages/dask/base.py:563: in compute
    results = schedule(dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:528: in get_sync
    return get_async(apply_sync, 1, dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:495: in get_async
    fire_task()
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:457: in fire_task
    apply_async(
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:517: in apply_sync
    res = func(*args, **kwds)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:227: in execute_task
    result = pack_exception(e, dumps)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:222: in execute_task
    result = _execute_task(task, data)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/optimization.py:963: in __call__
    return core.get(self.dsk, self.outkey, dict(zip(self.inkeys, args)))
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:151: in get
    result = _execute_task(task, cache)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/utils.py:35: in apply
    return func(*args, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/dataframe/core.py:5471: in apply_and_enforce
    df = func(*args, **kwargs)
nvtabular/workflow.py:336: in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
nvtabular/ops/fill.py:98: in transform
    df[col] = df[col].fillna(self.medians[col])
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py:1874: in fillna
    return super().fillna(
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py:1475: in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <cudf.core.column.numerical.NumericalColumn object at 0x7ff02829f4c0>
fill_value = 1000.5, method = None, dtype = None, fill_nan = True

    def fillna(
        self,
        fill_value: Any = None,
        method: str = None,
        dtype: Dtype = None,
        fill_nan: bool = True,
    ) -> NumericalColumn:
        """
        Fill null values with *fill_value*
        """
        if fill_nan:
            col = self.nans_to_nulls()
        else:
            col = self
    
        if method is not None:
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if fill_value is None:
            raise ValueError("Must specify either 'fill_value' or 'method'")
    
        if (
            isinstance(fill_value, cudf.Scalar)
            and fill_value.dtype == col.dtype
        ):
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if np.isscalar(fill_value):
            # cast safely to the same dtype as self
            fill_value_casted = col.dtype.type(fill_value)
            if not np.isnan(fill_value) and (fill_value_casted != fill_value):
>               raise TypeError(
                    f"Cannot safely cast non-equivalent "
                    f"{type(fill_value).__name__} to {col.dtype.name}"
                )
E               TypeError: Cannot safely cast non-equivalent float to int64

/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py:487: TypeError
------------------------------ Captured log call -------------------------------
ERROR    nvtabular:workflow.py:338 Failed to transform operator <nvtabular.ops.fill.FillMedian object at 0x7ff1059ff790>
Traceback (most recent call last):
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/workflow.py", line 336, in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
  File "/conda/envs/rapids/lib/python3.8/contextlib.py", line 75, in inner
    return func(*args, **kwds)
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/ops/fill.py", line 98, in transform
    df[col] = df[col].fillna(self.medians[col])
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py", line 1874, in fillna
    return super().fillna(
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py", line 1475, in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py", line 487, in fillna
    raise TypeError(
TypeError: Cannot safely cast non-equivalent float to int64
_______________________ test_gpu_dl[0-parquet-1000-0.06] _______________________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-55/test_gpu_dl_0_parquet_1000_0_01')
df =       name-cat name-string    id  label         x         y
0        Quinn        Gary  1020    996  0.672772 -0.59399...t  1027   1048  0.668489 -0.437601
4320    Xavier     Michael   996   1040  0.471682  0.617638

[4321 rows x 6 columns]
dataset = <nvtabular.io.dataset.Dataset object at 0x7ff1059fffd0>
batch_size = 1000, part_mem_fraction = 0.06, engine = 'parquet', device = 0

    @pytest.mark.parametrize("part_mem_fraction", [0.001, 0.06])
    @pytest.mark.parametrize("batch_size", [1000])
    @pytest.mark.parametrize("engine", ["parquet"])
    @pytest.mark.parametrize("device", [None, 0])
    def test_gpu_dl(tmpdir, df, dataset, batch_size, part_mem_fraction, engine, device):
        cat_names = ["name-cat", "name-string"]
        cont_names = ["x", "y", "id"]
        label_name = ["label"]
    
        conts = cont_names >> ops.FillMedian() >> ops.Normalize()
        cats = cat_names >> ops.Categorify()
    
        processor = nvt.Workflow(conts + cats + label_name)
    
        output_train = os.path.join(tmpdir, "train/")
        os.mkdir(output_train)
    
>       processor.fit_transform(dataset).to_parquet(
            shuffle=nvt.io.Shuffle.PER_PARTITION,
            output_path=output_train,
            out_files_per_proc=2,
        )

tests/unit/test_torch_dataloader.py:169: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:177: in fit_transform
    self.fit(dataset)
nvtabular/workflow.py:144: in fit
    results = dask.compute(stats, scheduler="synchronous")[0]
/conda/envs/rapids/lib/python3.8/site-packages/dask/base.py:563: in compute
    results = schedule(dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:528: in get_sync
    return get_async(apply_sync, 1, dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:495: in get_async
    fire_task()
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:457: in fire_task
    apply_async(
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:517: in apply_sync
    res = func(*args, **kwds)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:227: in execute_task
    result = pack_exception(e, dumps)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:222: in execute_task
    result = _execute_task(task, data)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/optimization.py:963: in __call__
    return core.get(self.dsk, self.outkey, dict(zip(self.inkeys, args)))
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:151: in get
    result = _execute_task(task, cache)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/utils.py:35: in apply
    return func(*args, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/dataframe/core.py:5471: in apply_and_enforce
    df = func(*args, **kwargs)
nvtabular/workflow.py:336: in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
nvtabular/ops/fill.py:98: in transform
    df[col] = df[col].fillna(self.medians[col])
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py:1874: in fillna
    return super().fillna(
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py:1475: in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <cudf.core.column.numerical.NumericalColumn object at 0x7ff028091ec0>
fill_value = 1000.5, method = None, dtype = None, fill_nan = True

    def fillna(
        self,
        fill_value: Any = None,
        method: str = None,
        dtype: Dtype = None,
        fill_nan: bool = True,
    ) -> NumericalColumn:
        """
        Fill null values with *fill_value*
        """
        if fill_nan:
            col = self.nans_to_nulls()
        else:
            col = self
    
        if method is not None:
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if fill_value is None:
            raise ValueError("Must specify either 'fill_value' or 'method'")
    
        if (
            isinstance(fill_value, cudf.Scalar)
            and fill_value.dtype == col.dtype
        ):
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if np.isscalar(fill_value):
            # cast safely to the same dtype as self
            fill_value_casted = col.dtype.type(fill_value)
            if not np.isnan(fill_value) and (fill_value_casted != fill_value):
>               raise TypeError(
                    f"Cannot safely cast non-equivalent "
                    f"{type(fill_value).__name__} to {col.dtype.name}"
                )
E               TypeError: Cannot safely cast non-equivalent float to int64

/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py:487: TypeError
------------------------------ Captured log call -------------------------------
ERROR    nvtabular:workflow.py:338 Failed to transform operator <nvtabular.ops.fill.FillMedian object at 0x7ff1059cfc40>
Traceback (most recent call last):
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/workflow.py", line 336, in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
  File "/conda/envs/rapids/lib/python3.8/contextlib.py", line 75, in inner
    return func(*args, **kwds)
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/ops/fill.py", line 98, in transform
    df[col] = df[col].fillna(self.medians[col])
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py", line 1874, in fillna
    return super().fillna(
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py", line 1475, in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py", line 487, in fillna
    raise TypeError(
TypeError: Cannot safely cast non-equivalent float to int64
_________________________ test_kill_dl[parquet-0.001] __________________________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-55/test_kill_dl_parquet_0_001_0')
df =       name-cat name-string    id  label         x         y
0        Quinn        Gary  1020    996  0.672772 -0.59399...t  1027   1048  0.668489 -0.437601
4320    Xavier     Michael   996   1040  0.471682  0.617638

[4321 rows x 6 columns]
dataset = <nvtabular.io.dataset.Dataset object at 0x7ff0185bccd0>
part_mem_fraction = 0.001, engine = 'parquet'

    @pytest.mark.parametrize("part_mem_fraction", [0.001, 0.1])
    @pytest.mark.parametrize("engine", ["parquet"])
    def test_kill_dl(tmpdir, df, dataset, part_mem_fraction, engine):
        cat_names = ["name-cat", "name-string"]
        cont_names = ["x", "y", "id"]
        label_name = ["label"]
    
        conts = cont_names >> ops.FillMedian() >> ops.Normalize()
        cats = cat_names >> ops.Categorify()
    
        processor = nvt.Workflow(conts + cats + label_name)
    
        output_train = os.path.join(tmpdir, "train/")
        os.mkdir(output_train)
    
>       processor.fit_transform(dataset).to_parquet(
            shuffle=nvt.io.Shuffle.PER_PARTITION,
            output_path=output_train,
            out_files_per_proc=2,
        )

tests/unit/test_torch_dataloader.py:238: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:177: in fit_transform
    self.fit(dataset)
nvtabular/workflow.py:144: in fit
    results = dask.compute(stats, scheduler="synchronous")[0]
/conda/envs/rapids/lib/python3.8/site-packages/dask/base.py:563: in compute
    results = schedule(dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:528: in get_sync
    return get_async(apply_sync, 1, dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:495: in get_async
    fire_task()
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:457: in fire_task
    apply_async(
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:517: in apply_sync
    res = func(*args, **kwds)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:227: in execute_task
    result = pack_exception(e, dumps)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:222: in execute_task
    result = _execute_task(task, data)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/optimization.py:963: in __call__
    return core.get(self.dsk, self.outkey, dict(zip(self.inkeys, args)))
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:151: in get
    result = _execute_task(task, cache)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/utils.py:35: in apply
    return func(*args, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/dataframe/core.py:5471: in apply_and_enforce
    df = func(*args, **kwargs)
nvtabular/workflow.py:336: in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
nvtabular/ops/fill.py:98: in transform
    df[col] = df[col].fillna(self.medians[col])
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py:1874: in fillna
    return super().fillna(
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py:1475: in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <cudf.core.column.numerical.NumericalColumn object at 0x7ff0522cce40>
fill_value = 1000.5, method = None, dtype = None, fill_nan = True

    def fillna(
        self,
        fill_value: Any = None,
        method: str = None,
        dtype: Dtype = None,
        fill_nan: bool = True,
    ) -> NumericalColumn:
        """
        Fill null values with *fill_value*
        """
        if fill_nan:
            col = self.nans_to_nulls()
        else:
            col = self
    
        if method is not None:
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if fill_value is None:
            raise ValueError("Must specify either 'fill_value' or 'method'")
    
        if (
            isinstance(fill_value, cudf.Scalar)
            and fill_value.dtype == col.dtype
        ):
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if np.isscalar(fill_value):
            # cast safely to the same dtype as self
            fill_value_casted = col.dtype.type(fill_value)
            if not np.isnan(fill_value) and (fill_value_casted != fill_value):
>               raise TypeError(
                    f"Cannot safely cast non-equivalent "
                    f"{type(fill_value).__name__} to {col.dtype.name}"
                )
E               TypeError: Cannot safely cast non-equivalent float to int64

/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py:487: TypeError
------------------------------ Captured log call -------------------------------
ERROR    nvtabular:workflow.py:338 Failed to transform operator <nvtabular.ops.fill.FillMedian object at 0x7ff0185bca00>
Traceback (most recent call last):
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/workflow.py", line 336, in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
  File "/conda/envs/rapids/lib/python3.8/contextlib.py", line 75, in inner
    return func(*args, **kwds)
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/ops/fill.py", line 98, in transform
    df[col] = df[col].fillna(self.medians[col])
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py", line 1874, in fillna
    return super().fillna(
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py", line 1475, in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py", line 487, in fillna
    raise TypeError(
TypeError: Cannot safely cast non-equivalent float to int64
__________________________ test_kill_dl[parquet-0.1] ___________________________

tmpdir = local('/tmp/pytest-of-jenkins/pytest-55/test_kill_dl_parquet_0_1_0')
df =       name-cat name-string    id  label         x         y
0        Quinn        Gary  1020    996  0.672772 -0.59399...t  1027   1048  0.668489 -0.437601
4320    Xavier     Michael   996   1040  0.471682  0.617638

[4321 rows x 6 columns]
dataset = <nvtabular.io.dataset.Dataset object at 0x7ff105f735e0>
part_mem_fraction = 0.1, engine = 'parquet'

    @pytest.mark.parametrize("part_mem_fraction", [0.001, 0.1])
    @pytest.mark.parametrize("engine", ["parquet"])
    def test_kill_dl(tmpdir, df, dataset, part_mem_fraction, engine):
        cat_names = ["name-cat", "name-string"]
        cont_names = ["x", "y", "id"]
        label_name = ["label"]
    
        conts = cont_names >> ops.FillMedian() >> ops.Normalize()
        cats = cat_names >> ops.Categorify()
    
        processor = nvt.Workflow(conts + cats + label_name)
    
        output_train = os.path.join(tmpdir, "train/")
        os.mkdir(output_train)
    
>       processor.fit_transform(dataset).to_parquet(
            shuffle=nvt.io.Shuffle.PER_PARTITION,
            output_path=output_train,
            out_files_per_proc=2,
        )

tests/unit/test_torch_dataloader.py:238: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
nvtabular/workflow.py:177: in fit_transform
    self.fit(dataset)
nvtabular/workflow.py:144: in fit
    results = dask.compute(stats, scheduler="synchronous")[0]
/conda/envs/rapids/lib/python3.8/site-packages/dask/base.py:563: in compute
    results = schedule(dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:528: in get_sync
    return get_async(apply_sync, 1, dsk, keys, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:495: in get_async
    fire_task()
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:457: in fire_task
    apply_async(
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:517: in apply_sync
    res = func(*args, **kwds)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:227: in execute_task
    result = pack_exception(e, dumps)
/conda/envs/rapids/lib/python3.8/site-packages/dask/local.py:222: in execute_task
    result = _execute_task(task, data)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/optimization.py:963: in __call__
    return core.get(self.dsk, self.outkey, dict(zip(self.inkeys, args)))
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:151: in get
    result = _execute_task(task, cache)
/conda/envs/rapids/lib/python3.8/site-packages/dask/core.py:121: in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
/conda/envs/rapids/lib/python3.8/site-packages/dask/utils.py:35: in apply
    return func(*args, **kwargs)
/conda/envs/rapids/lib/python3.8/site-packages/dask/dataframe/core.py:5471: in apply_and_enforce
    df = func(*args, **kwargs)
nvtabular/workflow.py:336: in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
/conda/envs/rapids/lib/python3.8/contextlib.py:75: in inner
    return func(*args, **kwds)
nvtabular/ops/fill.py:98: in transform
    df[col] = df[col].fillna(self.medians[col])
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py:1874: in fillna
    return super().fillna(
/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py:1475: in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <cudf.core.column.numerical.NumericalColumn object at 0x7ff031386740>
fill_value = 1000.5, method = None, dtype = None, fill_nan = True

    def fillna(
        self,
        fill_value: Any = None,
        method: str = None,
        dtype: Dtype = None,
        fill_nan: bool = True,
    ) -> NumericalColumn:
        """
        Fill null values with *fill_value*
        """
        if fill_nan:
            col = self.nans_to_nulls()
        else:
            col = self
    
        if method is not None:
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if fill_value is None:
            raise ValueError("Must specify either 'fill_value' or 'method'")
    
        if (
            isinstance(fill_value, cudf.Scalar)
            and fill_value.dtype == col.dtype
        ):
            return super(NumericalColumn, col).fillna(fill_value, method)
    
        if np.isscalar(fill_value):
            # cast safely to the same dtype as self
            fill_value_casted = col.dtype.type(fill_value)
            if not np.isnan(fill_value) and (fill_value_casted != fill_value):
>               raise TypeError(
                    f"Cannot safely cast non-equivalent "
                    f"{type(fill_value).__name__} to {col.dtype.name}"
                )
E               TypeError: Cannot safely cast non-equivalent float to int64

/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py:487: TypeError
------------------------------ Captured log call -------------------------------
ERROR    nvtabular:workflow.py:338 Failed to transform operator <nvtabular.ops.fill.FillMedian object at 0x7ff105f73670>
Traceback (most recent call last):
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/workflow.py", line 336, in _transform_partition
    df = column_group.op.transform(column_group.input_column_names, df)
  File "/conda/envs/rapids/lib/python3.8/contextlib.py", line 75, in inner
    return func(*args, **kwds)
  File "/var/jenkins_home/workspace/nvtabular_tests/nvtabular/nvtabular/ops/fill.py", line 98, in transform
    df[col] = df[col].fillna(self.medians[col])
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/series.py", line 1874, in fillna
    return super().fillna(
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/frame.py", line 1475, in fillna
    copy_data[name] = copy_data[name].fillna(value[name], method)
  File "/conda/envs/rapids/lib/python3.8/site-packages/cudf/core/column/numerical.py", line 487, in fillna
    raise TypeError(
TypeError: Cannot safely cast non-equivalent float to int64

----------- coverage: platform linux, python 3.8.8-final-0 -----------
Name                                                           Stmts   Miss Branch BrPart  Cover   Missing
----------------------------------------------------------------------------------------------------------
nvtabular/__init__.py                                             12      0      0      0   100%
nvtabular/column_group.py                                        149     18     80      5    86%   54, 87, 128, 151-164, 191, 278
nvtabular/dispatch.py                                             81     11     38      5    83%   35, 45->47, 69, 94, 111, 118, 135-138, 167-170
nvtabular/framework_utils/__init__.py                              0      0      0      0   100%
nvtabular/framework_utils/tensorflow/__init__.py                   1      0      0      0   100%
nvtabular/framework_utils/tensorflow/feature_column_utils.py     146    137     96      0     4%   28-32, 69-303
nvtabular/framework_utils/tensorflow/layers/__init__.py            4      0      0      0   100%
nvtabular/framework_utils/tensorflow/layers/embedding.py         153     14     89      7    87%   56, 64->45, 100, 108, 188, 240-248, 251, 344->352, 366->369, 372-373, 376
nvtabular/framework_utils/tensorflow/layers/interaction.py        47     25     20      1    43%   49, 74-103, 106-110, 113
nvtabular/framework_utils/tensorflow/layers/outer_product.py      30     24     10      0    15%   37-38, 41-60, 71-84, 87
nvtabular/framework_utils/torch/__init__.py                        0      0      0      0   100%
nvtabular/framework_utils/torch/layers/__init__.py                 2      0      0      0   100%
nvtabular/framework_utils/torch/layers/embeddings.py              27     12     12      1    51%   47, 71-79, 82-88
nvtabular/framework_utils/torch/models.py                         41      6     22      7    76%   56, 59, 64, 85, 87-88, 91->94, 98->100
nvtabular/framework_utils/torch/utils.py                          31      7     10      3    76%   52, 56-58, 67-69
nvtabular/inference/__init__.py                                    0      0      0      0   100%
nvtabular/inference/triton/__init__.py                           218    211     90      0     2%   24-494
nvtabular/inference/triton/model.py                               56     56     22      0     0%   27-142
nvtabular/inference/triton/model_hugectr.py                       44     44     14      0     0%   27-120
nvtabular/io/__init__.py                                           4      0      0      0   100%
nvtabular/io/avro.py                                              88     88     30      0     0%   16-189
nvtabular/io/csv.py                                               54      4     20      5    88%   95, 99->103, 104, 106, 120
nvtabular/io/dask.py                                             178      7     68     11    93%   109, 112, 148, 223, 380->378, 408->411, 419, 423->425, 425->421, 430, 432
nvtabular/io/dataframe_engine.py                                  58      3     30      6    90%   44, 63, 82->86, 86->91, 88->91, 91->110, 119
nvtabular/io/dataset.py                                          247     31    114     21    84%   254, 256, 269, 278, 296-310, 409->478, 414-417, 422->432, 427-428, 439->437, 453->457, 468, 507, 608->610, 610->619, 620, 627-628, 634, 640, 735-736, 848-853, 859, 884
nvtabular/io/dataset_engine.py                                    23      1      0      0    96%   45
nvtabular/io/hugectr.py                                           45      2     24      2    91%   34, 74->97, 101
nvtabular/io/parquet.py                                          472     28    150     13    93%   83-91, 136-143, 155, 184-186, 311-316, 354-359, 475->482, 545->550, 551-552, 672, 676, 680, 718, 735, 739, 746->748, 866->871, 876->886, 913
nvtabular/io/shuffle.py                                           30      7     12      2    69%   41-48
nvtabular/io/writer.py                                           169     12     64      5    92%   31, 48, 76, 122, 125, 202, 211, 214, 257, 278-280
nvtabular/io/writer_factory.py                                    18      2      8      2    85%   35, 60
nvtabular/loader/__init__.py                                       0      0      0      0   100%
nvtabular/loader/backend.py                                      280     28    114     13    89%   79, 84-85, 113, 130, 137-138, 142-146, 149, 211, 217-221, 227-231, 241, 247-248, 266-267, 306->310, 381, 385-386, 480, 487
nvtabular/loader/tensorflow.py                                   119     12     52      8    87%   52, 60-63, 73, 83, 286, 293, 301-303, 313->317, 346
nvtabular/loader/tf_utils.py                                      55     10     20      5    80%   29->32, 32->34, 39->41, 43, 50-51, 58-60, 66-70
nvtabular/loader/torch.py                                         43     10      8      0    69%   25-27, 30-36
nvtabular/ops/__init__.py                                         19      0      0      0   100%
nvtabular/ops/bucketize.py                                        25      4     16      2    76%   46, 49-52
nvtabular/ops/categorify.py                                      513     69    298     47    83%   223, 240, 244, 252, 260, 262, 284, 303-304, 336-337, 399-401, 459-461, 466->468, 541, 579, 608->611, 612-614, 621-622, 635-637, 638->606, 654, 664, 666, 672, 688-689, 694, 697->700, 710, 734-736, 739, 741->743, 755-758, 784, 788, 790, 802-805, 920, 922, 964->985, 970->985, 986-991, 1028, 1044->1049, 1048, 1058->1055, 1063->1055, 1071, 1079-1089
nvtabular/ops/clip.py                                             19      2      6      3    80%   45, 53->55, 56
nvtabular/ops/column_similarity.py                                88     22     32      5    69%   84, 154-155, 164-166, 174-190, 205->215, 207->210, 211, 221
nvtabular/ops/data_stats.py                                       57      1     24      4    94%   88->90, 90->92, 93->84, 96->84, 101
nvtabular/ops/difference_lag.py                                   26      0      8      1    97%   67->69
nvtabular/ops/dropna.py                                            9      0      0      0   100%
nvtabular/ops/fill.py                                             58      2     20      1    96%   93, 119
nvtabular/ops/filter.py                                           21      1      6      1    93%   44
nvtabular/ops/hash_bucket.py                                      32      2     18      2    88%   73, 102
nvtabular/ops/hashed_cross.py                                     29      3     13      4    83%   51, 66, 80->exit, 81
nvtabular/ops/join_external.py                                    69      5     28      6    89%   96, 98, 116, 129->133, 153, 168
nvtabular/ops/join_groupby.py                                     80      5     28      2    94%   106, 109->116, 183-184, 187-188
nvtabular/ops/lambdaop.py                                         27      3     10      3    84%   61, 65, 78
nvtabular/ops/logop.py                                             9      0      0      0   100%
nvtabular/ops/moments.py                                          65      0     20      0   100%
nvtabular/ops/normalize.py                                        64      6     14      2    87%   61->60, 67-68, 100-101, 123-124
nvtabular/ops/operator.py                                         15      1      2      1    88%   24
nvtabular/ops/rename.py                                           18      3     10      3    71%   41, 54, 58
nvtabular/ops/stat_operator.py                                    11      0      0      0   100%
nvtabular/ops/target_encoding.py                                 151     11     66      5    91%   143, 163->167, 170->179, 222-223, 226-227, 236-242, 335->338
nvtabular/tools/__init__.py                                        0      0      0      0   100%
nvtabular/tools/data_gen.py                                      235      1     62      4    98%   183->185, 230->234, 320->319, 322
nvtabular/tools/dataset_inspector.py                              49      9     18      0    75%   29-38
nvtabular/tools/inspector_script.py                               46     46      0      0     0%   17-168
nvtabular/utils.py                                                42     16     18      7    55%   20-21, 25-26, 35, 39, 48-51, 53-55, 58, 61, 67, 73
nvtabular/worker.py                                               68      1     30      2    97%   73, 83->98
nvtabular/workflow.py                                            140      9     62      4    93%   39, 125, 137-139, 242, 270-271, 341
----------------------------------------------------------------------------------------------------------
TOTAL                                                           4910   1032   2026    231    76%
Coverage XML written to file coverage.xml

Required test coverage of 70% reached. Total coverage: 76.23%
=========================== short test summary info ============================
FAILED tests/unit/test_tf_dataloader.py::test_tf_gpu_dl[True-1-parquet-0.01]
FAILED tests/unit/test_tf_dataloader.py::test_tf_gpu_dl[True-1-parquet-0.06]
FAILED tests/unit/test_tf_dataloader.py::test_tf_gpu_dl[True-10-parquet-0.01]
FAILED tests/unit/test_tf_dataloader.py::test_tf_gpu_dl[True-10-parquet-0.06]
FAILED tests/unit/test_tf_dataloader.py::test_tf_gpu_dl[True-100-parquet-0.01]
FAILED tests/unit/test_tf_dataloader.py::test_tf_gpu_dl[True-100-parquet-0.06]
FAILED tests/unit/test_tf_dataloader.py::test_tf_gpu_dl[False-1-parquet-0.01]
FAILED tests/unit/test_tf_dataloader.py::test_tf_gpu_dl[False-1-parquet-0.06]
FAILED tests/unit/test_tf_dataloader.py::test_tf_gpu_dl[False-10-parquet-0.01]
FAILED tests/unit/test_tf_dataloader.py::test_tf_gpu_dl[False-10-parquet-0.06]
FAILED tests/unit/test_tf_dataloader.py::test_tf_gpu_dl[False-100-parquet-0.01]
FAILED tests/unit/test_tf_dataloader.py::test_tf_gpu_dl[False-100-parquet-0.06]
FAILED tests/unit/test_torch_dataloader.py::test_empty_cols[label_name0-cont_names0-cat_names0-parquet]
FAILED tests/unit/test_torch_dataloader.py::test_empty_cols[label_name0-cont_names0-cat_names1-parquet]
FAILED tests/unit/test_torch_dataloader.py::test_empty_cols[label_name1-cont_names0-cat_names0-parquet]
FAILED tests/unit/test_torch_dataloader.py::test_empty_cols[label_name1-cont_names0-cat_names1-parquet]
FAILED tests/unit/test_torch_dataloader.py::test_gpu_dl_break[None-parquet-1000-0.001]
FAILED tests/unit/test_torch_dataloader.py::test_gpu_dl_break[None-parquet-1000-0.06]
FAILED tests/unit/test_torch_dataloader.py::test_gpu_dl_break[0-parquet-1000-0.001]
FAILED tests/unit/test_torch_dataloader.py::test_gpu_dl_break[0-parquet-1000-0.06]
FAILED tests/unit/test_torch_dataloader.py::test_gpu_dl[None-parquet-1000-0.001]
FAILED tests/unit/test_torch_dataloader.py::test_gpu_dl[None-parquet-1000-0.06]
FAILED tests/unit/test_torch_dataloader.py::test_gpu_dl[0-parquet-1000-0.001]
FAILED tests/unit/test_torch_dataloader.py::test_gpu_dl[0-parquet-1000-0.06]
FAILED tests/unit/test_torch_dataloader.py::test_kill_dl[parquet-0.001] - Typ...
FAILED tests/unit/test_torch_dataloader.py::test_kill_dl[parquet-0.1] - TypeE...
============ 26 failed, 729 passed, 14 skipped in 490.99s (0:08:10) ============
Build step 'Execute shell' marked build as failure
Performing Post build task...
Match found for : : True
Logical operation result is TRUE
Running script  : #!/bin/bash
source activate rapids
cd /var/jenkins_home/
CUDA_VISIBLE_DEVICES=1 python test_res_push.py "https://api.GitHub.com/repos/NVIDIA/NVTabular/issues/$ghprbPullId/comments" "/var/jenkins_home/jobs/$JOB_NAME/builds/$BUILD_NUMBER/log" 
[nvtabular_tests] $ /bin/bash /tmp/jenkins292754657294673574.sh
</pre></details>

Fix patch:
diff --git a/nvtabular/ops/__init__.py b/nvtabular/ops/__init__.py
index d57f84a7b89..ac58f51352a 100644
--- a/nvtabular/ops/__init__.py
+++ b/nvtabular/ops/__init__.py
@@ -24,6 +24,7 @@
 from .dropna import Dropna
 from .fill import FillMedian, FillMissing
 from .filter import Filter
+from .groupby import Groupby
 from .hash_bucket import HashBucket
 from .hashed_cross import HashedCross
 from .join_external import JoinExternal
diff --git a/nvtabular/ops/groupby.py b/nvtabular/ops/groupby.py
new file mode 100644
index 00000000000..ce7a48cd196
--- /dev/null
+++ b/nvtabular/ops/groupby.py
@@ -0,0 +1,229 @@
+#
+# Copyright (c) 2021, 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.
+from collections import defaultdict
+
+import cudf
+from dask.dataframe.utils import meta_nonempty
+from nvtx import annotate
+
+from .operator import ColumnNames, Operator
+
+
+class Groupby(Operator):
+    """Groupby Transformation
+
+    Locally transform each partition of a Dataset with one or
+    more groupby aggregations.
+
+    WARNING: This transformation does NOT move data between
+    partitions. Please make sure that the target Dataset object
+    is already shuffled by ``groupby_cols``, otherwise the
+    output may be incorrect. See: ``Dataset.shuffle_by_keys``.
+
+    Example usage::
+
+        groupby_features = [
+            'user_id', 'session_id', 'month', 'prod_id',
+        ] >> ops.Groupby(
+            groupby_cols=['user_id', 'session_id'],
+            sort_cols=['month'],
+            aggs={
+                'prod_id': 'list',
+                'month': ['first', 'last'],
+            },
+        )
+        processor = nvtabular.Workflow(groupby_features)
+
+    Parameters
+    -----------
+    groupby_cols : str or list of str
+        The column names to be used as groupby keys.
+    sort_cols : str or list of str
+        Columns to be used to sort each partition before
+        groupby aggregation is performed. If this argument
+        is not specified, the results will not be sorted.
+    aggs : dict, list or str
+        Groupby aggregations to perform. Supported list-based
+        aggregations include "list", "first" & "last". Most
+        conventional aggregations supported by Pandas/cuDF are
+        also allowed (e.g. "sum", "count", "max", "mean", etc.).
+    name_sep : str
+        String separator to use for new column names.
+    """
+
+    def __init__(self, groupby_cols=None, sort_cols=None, aggs="list", name_sep="_"):
+        self.groupby_cols = groupby_cols
+        self.sort_cols = sort_cols or []
+        if isinstance(self.groupby_cols, str):
+            self.groupby_cols = [self.groupby_cols]
+        if isinstance(self.sort_cols, str):
+            self.sort_cols = [self.sort_cols]
+
+        # Split aggregations into "conventional" aggregations
+        # and "list-based" aggregations.  After this block,
+        # we will have a dictionary for each of these cases.
+        # We use the "__all__" key to specify aggregations
+        # that will be performed on all (non-key) columns.
+        self.list_aggs = defaultdict(list)
+        self.conv_aggs = defaultdict(list)
+        if isinstance(aggs, str):
+            aggs = {"__all__": [aggs]}
+        elif isinstance(aggs, list):
+            aggs = {"__all__": aggs}
+        for col, v in aggs.items():
+            _aggs = v if isinstance(v, list) else [v]
+            for _agg in _aggs:
+                if _is_list_agg(_agg):
+                    self.list_aggs[col].append(_agg)
+                else:
+                    self.conv_aggs[col].append(_agg)
+
+        self.name_sep = name_sep
+        super().__init__()
+
+    @annotate("Groupby_op", color="darkgreen", domain="nvt_python")
+    def transform(self, columns: ColumnNames, gdf: cudf.DataFrame) -> cudf.DataFrame:
+
+        # Sort if necessary
+        if self.sort_cols:
+            gdf = gdf.sort_values(self.sort_cols, ignore_index=True)
+
+        # List aggregations do not work with empty data.
+        # Use synthetic metadata to predict output columns.
+        empty_df = not len(gdf)
+        _df = meta_nonempty(gdf) if empty_df else gdf
+
+        # Get "complete" aggregation dicts
+        _list_aggs, _conv_aggs = _get_agg_dicts(
+            self.groupby_cols, self.list_aggs, self.conv_aggs, columns
+        )
+
+        # Get list-aggregation result
+        df_la = _apply_list_aggs(_df, self.groupby_cols, _list_aggs, name_sep=self.name_sep)
+
+        # Get conventional-aggregation result
+        df_cv = _apply_conventional_aggs(_df, self.groupby_cols, _conv_aggs, name_sep=self.name_sep)
+
+        if df_cv is None:
+            # Only using list aggregations
+            new_gdf = df_la
+        elif df_la is None:
+            # Only using conventional aggregations
+            new_gdf = df_cv
+        else:
+            # Using both aggregations, merge results
+            new_gdf = df_cv.merge(df_la, self.groupby_cols, how="outer")
+
+        if empty_df:
+            return new_gdf.iloc[:0]
+        return new_gdf
+
+    transform.__doc__ = Operator.transform.__doc__
+
+    def output_column_names(self, columns):
+        # Get the expected column names after transformation
+        _list_aggs, _conv_aggs = _get_agg_dicts(
+            self.groupby_cols, self.list_aggs, self.conv_aggs, columns
+        )
+        _list_aggs = _columns_out_from_aggs(_list_aggs, name_sep=self.name_sep)
+        _conv_aggs = _columns_out_from_aggs(_conv_aggs, name_sep=self.name_sep)
+
+        return list(set(self.groupby_cols) | set(_list_aggs) | set(_conv_aggs))
+
+
+def _columns_out_from_aggs(aggs, name_sep="_"):
+    # Helper function for `output_column_names`
+    _agg_cols = []
+    for k, v in aggs.items():
+        for _v in v:
+            _agg_cols.append(name_sep.join([k, _v]))
+    return _agg_cols
+
+
+def _apply_conventional_aggs(_df, groupby_cols, _conv_aggs, name_sep="_"):
+    df_cv = None  # Default return
+    if _conv_aggs:
+        _columns = list(set(groupby_cols) | set(_conv_aggs))
+        df_cv = _df[_columns].groupby(groupby_cols).agg(_conv_aggs).reset_index()
+        df_cv.columns = [
+            name_sep.join([n for n in name if n != ""]) for name in df_cv.columns.to_flat_index()
+        ]
+    return df_cv
+
+
+def _apply_list_aggs(_df, groupby_cols, _list_aggs, name_sep="_"):
+    df_la = None  # Default return
+    if _list_aggs:
+
+        # Perform initial "collect" aggregation
+        _columns = list(set(groupby_cols) | set(_list_aggs))
+        df_la = _df[_columns].groupby(groupby_cols).collect().reset_index()
+        columns = [c + f"{name_sep}list" if c in _list_aggs else c for c in df_la.columns]
+        df_la.columns = columns
+
+        # Handle "first" and "last" aggregations
+        for col, aggs in _list_aggs.items():
+            for _agg in aggs:
+                if _agg in ("first", "last"):
+                    df_la[f"{col}{name_sep}{_agg}"] = _first_or_last(
+                        df_la[f"{col}{name_sep}list"], _agg
+                    )
+            if "list" not in aggs:
+                df_la.drop(columns=[col + f"{name_sep}list"], inplace=True)
+    return df_la
+
+
+def _get_agg_dicts(groupby_cols, list_aggs, conv_aggs, columns):
+    # Get updated aggregation dicts. This should map "__all__"
+    # to specific columns, and remove elements that are not
+    # in `columns`.
+    _allowed_cols = [c for c in columns if c not in groupby_cols]
+    _list_aggs = _ensure_agg_dict(list_aggs, _allowed_cols)
+    _conv_aggs = _ensure_agg_dict(conv_aggs, _allowed_cols)
+    return _list_aggs, _conv_aggs
+
+
+def _ensure_agg_dict(_aggs, _allowed_cols):
+    # Make sure aggregation dict has legal keys
+    if "__all__" in _aggs:
+        return {col: _aggs["__all__"] for col in _allowed_cols}
+    else:
+        return {k: v for k, v in _aggs.items() if k in _allowed_cols}
+
+
+def _is_list_agg(agg):
+    # check if `agg` is a supported list aggregation
+    return agg in ("list", "first", "last")
+
+
+def _first_or_last(x, kind):
+    # Redirect to _first or _last
+    return _first(x) if kind == "first" else _last(x)
+
+
+def _first(x):
+    # Convert each element of a list column to be the first
+    # item in the list
+    offsets = x.list._column.offsets
+    elements = x.list._column.elements
+    return elements[offsets[:-1]]
+
+
+def _last(x):
+    # Convert each element of a list column to be the last
+    # item in the list
+    offsets = x.list._column.offsets
+    elements = x.list._column.elements
+    return elements[offsets[1:].values - 1]


Test patch:
diff --git a/tests/unit/test_ops.py b/tests/unit/test_ops.py
index 437035af4b8..91f079f1e54 100644
--- a/tests/unit/test_ops.py
+++ b/tests/unit/test_ops.py
@@ -963,3 +963,56 @@ def test_data_stats(tmpdir, df, datasets, engine):
         assert output[col]["per_nan"] == pytest.approx(
             100 * (1 - ddf[col].count().compute() / len(ddf[col]))
         )
+
+
+@pytest.mark.parametrize("keys", [["name"], "id", ["name", "id"]])
+def test_groupby_op(keys):
+
+    # Initial timeseries dataset
+    size = 60
+    df1 = pd.DataFrame(
+        {
+            "name": np.random.choice(["Dave", "Zelda"], size=size),
+            "id": np.random.choice([0, 1], size=size),
+            "ts": np.linspace(0.0, 10.0, num=size),
+            "x": np.arange(size),
+            "y": np.linspace(0.0, 10.0, num=size),
+            "shuffle": np.random.uniform(low=0.0, high=10.0, size=size),
+        }
+    )
+    df1 = df1.sort_values("shuffle").drop(columns="shuffle").reset_index(drop=True)
+
+    # Create a ddf, and be sure to shuffle by the groupby keys
+    ddf1 = dd.from_pandas(df1, npartitions=3).shuffle(keys)
+    dataset = nvt.Dataset(ddf1)
+
+    # Define Groupby Workflow
+    groupby_features = ColumnGroup(["name", "id", "ts", "x", "y"]) >> ops.Groupby(
+        groupby_cols=keys,
+        sort_cols=["ts"],
+        aggs={
+            "x": ["list", "sum"],
+            "y": ["first", "last"],
+        },
+        name_sep="-",
+    )
+    processor = nvtabular.Workflow(groupby_features)
+    processor.fit(dataset)
+    new_gdf = processor.transform(dataset).to_ddf().compute()
+
+    # Check list-aggregation ordering
+    x = new_gdf["x-list"]
+    x = x.to_pandas() if hasattr(x, "to_pandas") else x
+    sums = []
+    for el in x.values:
+        _el = pd.Series(el)
+        sums.append(_el.sum())
+        assert _el.is_monotonic_increasing
+
+    # Check that list sums match sum aggregation
+    x = new_gdf["x-sum"]
+    x = x.to_pandas() if hasattr(x, "to_pandas") else x
+    assert list(x) == sums
+
+    # Check basic behavior or "y" column
+    assert (new_gdf["y-first"] < new_gdf["y-last"]).all()