Eager: Eager execution of tf.data pipelines

[hsm207]

System information

Tensorflow version:

1.5.0-dev20171120

Python version:

python 3.6.3 |Anaconda, Inc.| (default, Nov 8 2017, 15:10:56) [MSC v.1900 64 bit (AMD64)]

Problem

When debugging, calling the numpy() method on a Tensor object results in AttributeError: 'Tensor' object has no attribute 'numpy' in certain situations.

Steps to reproduce

1. Put this code in a script:

import tensorflow as tf
import tensorflow.contrib.eager as tfe
import numpy as np
import tensorflow as tf
from collections import defaultdict, Counter

tfe.enable_eager_execution()

class_probs = dict(
    a=0.15,
    b=0.3,
    c=0.8,
    d=0.9,
    e=0.2,
    f=0.02
)
num_classes = len(class_probs)

class_probs = {k: v / sum(class_probs.values()) for k, v in class_probs.items()}
class_mapping = {n: i for i, n in enumerate(class_probs.keys())}

class_names = list(class_probs.keys())
class_weights = list(class_probs.values())
sampled_dataset = np.random.choice(class_names, size=1000, p=class_weights)

dataset_data = defaultdict(list)
for i, d in enumerate(sampled_dataset):
    dataset_data['class_name'].append(d)
    dataset_data['class_id'].append(class_mapping[d])
    dataset_data['data'].append(np.array([i]))
    dataset_data['class_prob'].append(class_probs[d])

    dataset_data['class_target_prob'].append(1 / num_classes)

for k, v in dataset_data.items():
    dataset_data[k] = np.array(dataset_data[k])

class_counts = Counter(sampled_dataset)

oversampling_coef = 0.9


def oversample_classes(example):
    """
    Returns the number of copies of given example
    """
    class_prob = example['class_prob']
    class_target_prob = example['class_target_prob']
    prob_ratio = tf.cast(class_target_prob / class_prob, dtype=tf.float32)

    prob_ratio = prob_ratio ** oversampling_coef

    prob_ratio = tf.maximum(prob_ratio, 1)
    # Breakpoint 1
    repeat_count = tf.floor(prob_ratio)

    repeat_residual = prob_ratio - repeat_count  # a number between 0-1
    residual_acceptance = tf.less_equal(
        tf.random_uniform([], dtype=tf.float32), repeat_residual
    )

    residual_acceptance = tf.cast(residual_acceptance, tf.int64)
    repeat_count = tf.cast(repeat_count, dtype=tf.int64)

    return repeat_count + residual_acceptance


dataset = tf.data.Dataset.from_tensor_slices(dict(dataset_data))

dataset = dataset.flat_map(
    lambda x: tf.data.Dataset.from_tensors(x).repeat(oversample_classes(x))
)

i = tfe.Iterator(dataset)
x = i.next()['class_id']

# Breakpoint 2
print('end')

2. Insert the breakpoints in the lines following the comments Breakpoint 1 and Breakpoint 2

3. Debug the script

4. When breakpoint 1 is reached, evaluate the following:
   prob_ratio.numpy()
   This will result in the attribute error message.

5. When breakpoint 2 is reached, evaluate the following:
   x.numpy()
   This will not result in the attribute error message.

[hsm207]

@asimshankar Is there any workaround while waiting for this bug to be fixed?

[asimshankar]

Apologies for the late response, I was out. Long story short: This behavior is intended and it isn't a bug :)

The tf.data API defines input processing as a dataflow graph that is executed by the TensorFlow runtime. In the code sample above, the call to dataset.flat_map expects a function that constructs the dataflow graph defining the transformations to be performed. As a result, you'll observe that breakpoint 1 will be triggered by the call to dataset.flat_map and not by the call to i.next(). In a program that iterates over the dataset, you'll observe that the breakpoint is hit exactly once, and not once per call to i.next().

Expressing the input pipeline as a dataflow graph enables efficient background input processing - allowing multiple input tensors to be processed concurrently (e.g., multiple threads for tf.data.Dataset.map), and for input processing to be conducted concurrently with model computation. By doing so, input processing isn't bottlenecked on the single Python interpreter thread.

That said, a workaround would be to use something like a tf.py_func which will bring control to the Python interpreter. @akshayka is actively working on a tfe.py_func which will make this possible (at an efficiency cost of sharing the single Python interpreter thread).

Please let me know if the comment makes sense.

I'm going to keep the issue open and we can close it when tfe.py_func is in place and we can provide you with a sample of how to use it in your input pipeline.

FYI @mrry @jsimsa

[hsm207]

@asimshankar Thank you for the explanation. It makes total sense to me.

[MtDersvan]

@asimshankar, on a related note. Are lookup_ops supported inside the tf.data.Dataset.map in the eager mode?

I'm getting a similar Error when using lookup_ops in it:

ValueError: Tried to convert 'table_handle' to a tensor and failed. 
Error: Resource handles are not convertible to numpy.

in a .map() function:

tf_ds = tf_ds.map(parse_tfrecord_dataset)

while the cause of the exception are the following lines in the parse_tfrecord_dataset:

example = tf.cond(tf.equal(example, tf.convert_to_tensor('NA')),
                           lambda: tf.convert_to_tensor(-1, tf.int64),
                           lambda: example_table.lookup(example))

Are there any workarounds for including lookup tables into eager data pipelines?

[asimshankar]

@MtDersvan : Very sorry for the late reply, somehow missed this. This is related to (as you probably figured out #15212) - ops that have some implicit state stored in a DT_RESOURCE tensor need some work. We're looking into that.

[tensorflowbutler]

Nagging Assignee: It has been 14 days with no activity and this issue has an assignee. Please update the label and/or status accordingly.

[tensorflowbutler]

Nagging Assignee: It has been 14 days with no activity and this issue has an assignee. Please update the label and/or status accordingly.

[tensorflowbutler]

Nagging Assignee: It has been 14 days with no activity and this issue has an assignee. Please update the label and/or status accordingly.

[tensorflowbutler]

Nagging Assignees @asimshankar, @akshayka: It has been 14 days with no activity and this issue has an assignee. Please update the label and/or status accordingly.