With a tf.data pipeline, there are several spots where you can parallelize. Depending on how your data are stored and read, you can parallelize reading. You can also parallelize augmentation, and you can prefetch data as you train, so your GPU (or other hardware) is never hungry for data.

In the code below, I have demonstrated how you can parallelize augmentation and add prefetching.

import numpy as np
import tensorflow as tf

x_shape = (32, 32, 3)
y_shape = ()  # A single item (not array).
classes = 10

# This is tf.data.experimental.AUTOTUNE in older tensorflow.
AUTOTUNE = tf.data.AUTOTUNE

def generator_fn(n_samples):
    """Return a function that takes no arguments and returns a generator."""
    def generator():
        for i in range(n_samples):
            # Synthesize an image and a class label.
            x = np.random.random_sample(x_shape).astype(np.float32)
            y = np.random.randint(0, classes, size=y_shape, dtype=np.int32)
            yield x, y
    return generator

def augment(x, y):
    return x * tf.random.normal(shape=x_shape), y

samples = 10
batch_size = 5
epochs = 2

# Create dataset.
gen = generator_fn(n_samples=samples)
dataset = tf.data.Dataset.from_generator(
    generator=gen, 
    output_types=(np.float32, np.int32), 
    output_shapes=(x_shape, y_shape)
)
# Parallelize the augmentation.
dataset = dataset.map(
    augment, 
    num_parallel_calls=AUTOTUNE,
    # Order does not matter.
    deterministic=False
)
dataset = dataset.batch(batch_size, drop_remainder=True)
# Prefetch some batches.
dataset = dataset.prefetch(AUTOTUNE)

# Prepare model.
model = tf.keras.applications.VGG16(weights=None, input_shape=x_shape, classes=classes)
model.compile(optimizer="adam", loss="sparse_categorical_crossentropy")

# Train. Do not specify batch size because the dataset takes care of that.
model.fit(dataset, epochs=epochs)

Suppose you have a function that generates data:

 def generator(data): 
    ...
    yield (X, y)

Now you need another function that describes your model architecture. It could be any function that processes X and has to predict y as output (say, neural network).

Suppose your function accepts X and y as inputs, computes a prediction for y from X in some way and returns loss function (e.g. cross-entropy or MSE in the case of regression) between y and predicted y:

 def neural_network(X, y): 
    # computation of prediction for y using X
    ...
    return loss(y, y_pred)

To make your model work, you need to define placeholders for both X and y and then run a session:

 X = tf.placeholder(tf.float32, shape=(batch_size, x_dim))
 y = tf.placeholder(tf.float32, shape=(batch_size, y_dim))

Placeholders are something like "free variables" which you need to specify when running the session by feed_dict:

 with tf.Session() as sess:
     # variables need to be initialized before any sess.run() calls
     tf.global_variables_initializer().run()

     for X_batch, y_batch in generator(data):
         feed_dict = {X: X_batch, y: y_batch} 
         _, loss_value, ... = sess.run([train_op, loss, ...], feed_dict)
         # train_op here stands for optimization operation you have defined
         # and loss for loss function (return value of neural_network function)

Hope you would find it useful. However, bear in mind this is not fully working implementation but rather a pseudocode since you specified almost no details.

Custom Image Data Generator

load Directory data into dataframe for CustomDataGenerator

def data_to_df(data_dir, subset=None, validation_split=None):
    df = pd.DataFrame()
    filenames = []
    labels = []
    
    for dataset in os.listdir(data_dir):
        img_list = os.listdir(os.path.join(data_dir, dataset))
        label = name_to_idx[dataset]
        
        for image in img_list:
            filenames.append(os.path.join(data_dir, dataset, image))
            labels.append(label)
        
    df["filenames"] = filenames
    df["labels"] = labels
    
    if subset == "train":
        split_indexes = int(len(df) * validation_split)
        train_df = df[split_indexes:]
        val_df = df[:split_indexes]
        return train_df, val_df
    
    return df

train_df, val_df = data_to_df(train_dir, subset="train", validation_split=0.2)

Custom Data Generator

import tensorflow as tf
from PIL import Image
import numpy as np

class CustomDataGenerator(tf.keras.utils.Sequence):

    ''' Custom DataGenerator to load img 
    
    Arguments:
        data_frame = pandas data frame in filenames and labels format
        batch_size = divide data in batches
        shuffle = shuffle data before loading
        img_shape = image shape in (h, w, d) format
        augmentation = data augmentation to make model rebust to overfitting
    
    Output:
        Img: numpy array of image
        label : output label for image
    '''
    
    def __init__(self, data_frame, batch_size=10, img_shape=None, augmentation=True, num_classes=None):
        self.data_frame = data_frame
        self.train_len = len(data_frame)
        self.batch_size = batch_size
        self.img_shape = img_shape
        self.num_classes = num_classes
        print(f"Found {self.data_frame.shape[0]} images belonging to {self.num_classes} classes")

    def __len__(self):
        ''' return total number of batches '''
        self.data_frame = shuffle(self.data_frame)
        return math.ceil(self.train_len/self.batch_size)

    def on_epoch_end(self):
        ''' shuffle data after every epoch '''
        # fix on epoch end it's not working, adding shuffle in len for alternative
        pass
    
    def __data_augmentation(self, img):
        ''' function for apply some data augmentation '''
        img = tf.keras.preprocessing.image.random_shift(img, 0.2, 0.3)
        img = tf.image.random_flip_left_right(img)
        img = tf.image.random_flip_up_down(img)
        return img
        
    def __get_image(self, file_id):
        """ open image with file_id path and apply data augmentation """
        img = np.asarray(Image.open(file_id))
        img = np.resize(img, self.img_shape)
        img = self.__data_augmentation(img)
        img = preprocess_input(img)

        return img

    def __get_label(self, label_id):
        """ uncomment the below line to convert label into categorical format """
        #label_id = tf.keras.utils.to_categorical(label_id, num_classes)
        return label_id

    def __getitem__(self, idx):
        batch_x = self.data_frame["filenames"][idx * self.batch_size:(idx + 1) * self.batch_size]
        batch_y = self.data_frame["labels"][idx * self.batch_size:(idx + 1) * self.batch_size]
        # read your data here using the batch lists, batch_x and batch_y
        x = [self.__get_image(file_id) for file_id in batch_x] 
        y = [self.__get_label(label_id) for label_id in batch_y]

        return tf.convert_to_tensor(x), tf.convert_to_tensor(y)

The tf.data.Dataset.from_generator() method provides a way to convert Python generators into tf.Tensor objects that evaluate to each successive element from the generator.

Let's say you have a simple generator that generates tuples (but could alternatively generate lists or NumPy arrays):

def g():
  yield 1, 10.0, "foo"
  yield 2, 20.0, "bar"
  yield 3, 30.0, "baz"

You can use the tf.data API to convert the generator first to a tf.data.Dataset, then to a tf.data.Iterator, and finally to a tuple of tf.Tensor objects.

dataset = tf.data.Dataset.from_generator(g, (tf.int32, tf.float32, tf.string))

iterator = dataset.make_one_shot_iterator()

int_tensor, float_tensor, str_tensor = iterator.get_next()

You can then use int_tensor, float_tensor, and str_tensor as the inputs to your TensorFlow model. See the tf.data programmer's guide for more ideas.