Natville

Let’s Talk

TensorFlow Data Pipeline

Github Link

https://github.com/Natan-Asrat/tensorflow_data_pipeline

Contact

The Setup

Introduction

In this project, i explore TensorFlow data pipeline by accomplishing three tasks concerned with building data generators, image augmentation, and dataset generation and training.

Libraries Used

Imports

Python 
import tensorflow as tf
import os
import numpy as np
import pandas as pd
from tensorflow.keras import Sequential
from tensorflow.keras.datasets import cifar10
from tensorflow.keras import Model
from tensorflow.keras.layers import Dense, Input, BatchNormalization
from tensorflow.keras.preprocessing.image import ImageDataGenerator
from sklearn.preprocessing import LabelBinarizer

Data Generator

Load the fertility dataset

Python 
headers = ['Season', 'Age', 'Diseases', 'Trauma', 'Surgery', 'Fever', 'Alcohol', 'Smoking', 'Sitting', 'Output']
fertility = pd.read_csv('data/fertility_diagnosis.txt', delimiter=',', header=None, names=headers)

Process the Data

Map the 'Output' feature from 'N' to 0 and from 'O' to 1:
Python 
fertility['Output'] = fertility['Output'].map(lambda x : 0.0 if x=='N' else 1.0)

Convert the DataFrame so that the features are mapped to floats:

Python 
fertility = fertility.astype('float32')

Shuffle the DataFrame:

Python 
fertility = fertility.sample(frac=1).reset_index(drop=True)

Convert the field Season to a one-hot encoded vector:

Python 
fertility = pd.get_dummies(fertility, prefix='Season', columns=['Season'])

Move the Output column such that it is the last column in the DataFrame:

Python 
fertility = fertility.reindex(columns = [col for col in fertility.columns if col != 'Output'] + ['Output'])

Convert the DataFrame to a numpy array:

Python 
fertility = fertility.to_numpy()

Split the Data

Split the dataset into training and validation set:

Python 
training = fertility[0:70]
validation = fertility[70:100]

Separate the features and labels for the validation and training data:

Python 
training_features = training[:,0:-1]
training_labels = training[:,-1]
validation_features = validation[:,0:-1]
validation_labels = validation[:,-1]

Create the Generator

Python 
def get_generator(features, labels, batch_size=1):
    for n in range(int(len(features)/batch_size)):
        yield (features[n*batch_size: (n+1)*batch_size], labels[n*batch_size: (n+1)*batch_size])

Apply the function to our training features and labels with a batch size of 10:

Python 
train_generator = get_generator(training_features, training_labels, batch_size=10)

Test the generator using the next() function:

Python 
next(train_generator)

Build the Model

Create a model using Keras with 3 layers:

Python 
input_shape = (12,)
output_shape = (1,)

model_input = Input(input_shape)
batch_1 = BatchNormalization(momentum=0.8)(model_input)
dense_1 = Dense(100, activation='relu')(batch_1)
batch_2 = BatchNormalization(momentum=0.8)(dense_1)
output = Dense(1, activation='sigmoid')(batch_2)

model = Model([model_input], output)

Compile the Model

Create the optimizer object:

Python 
optimizer = tf.keras.optimizers.Adam(learning_rate=1e-2)

Compile the model with loss function and metric:

Python 
model.compile(optimizer=optimizer, loss='binary_crossentropy', metrics=['accuracy'])

Fit the Model

Calculate the number of training steps per epoch for the given batch size:

Python 
batch_size = 5
train_steps = len(training) // batch_size

Train the model:

Python 
epochs = 3
for epoch in range(epochs):
    train_generator=get_generator(training_features, training_labels, batch_size =batch_size)
    validation_generator =get_generator(validation_features,validation_labels, batch_size=batch_size)
    model.fit_generator(train_generator,steps_per_epoch= train_steps, validation_data =validation_generator, validation_steps=1)

Evaluate and get Predictions

Obtain a validation data generator:

Python 
validation_generator = get_generator(validation_features, validation_labels, batch_size=30)

Evaluate the model:

Python 
predictions = model.evaluate_generator(validation_generator, steps=1)

Image Data Augmentation

Load the CIFAR-10 Dataset:

Python 
(training_features, training_labels), (test_features, test_labels) = cifar10.load_data()

Convert the labels to a one-hot encoding:

Python 
num_classes = 10

training_labels = tf.keras.utils.to_categorical(training_labels, num_classes)
test_labels = tf.keras.utils.to_categorical(test_labels, num_classes)

Create a Generator Function

Python 
def get_generator(features, labels, batch_size=1):
    for n in range(int(len(features)/batch_size)):
        yield (features[n*batch_size:(n+1)*batch_size], labels[n*batch_size:(n+1)*batch_size])

Use the function we created to get a training data generator with a batch size of 1:

Python 
training_generator = get_generator(training_features, training_labels)

Create a Data Augmentation Generator

Create a function to convert an image to monochrome:

Python 
def monochrome(x):
    def func_bw(a):
        average_colour = np.mean(a)
        return [average_colour, average_colour, average_colour]
    x = np.apply_along_axis(func_bw, -1, x)
    return x

Create an ImageDataGenerator object:

Python 
image_generator = ImageDataGenerator(
preprocessing_function=monochrome, rotation_range=180, rescale=(1/255.0) )
image_generator.fit(training_features)

Create an iterable generator using the `flow` function:

Python 
image_generator_iterable = image_generator.flow(training_features, training_labels, batch_size=1, shuffle=False)

Flow from Directory

Python 
train_path = 'data/flowers-recognition-split/train'
val_path = 'data/flowers-recognition-split/val'

datagenerator = ImageDataGenerator(rescale=(1/255.0))
classes = ['daisy', 'dandelion', 'rose', 'sunflower', 'tulip']

train_generator = datagenerator.flow_from_directory(train_path, batch_size = 64, classes=classes, target_size=(16,16))
val_generator = datagenerator.flow_from_directory(val_path, batch_size = 64, classes=classes, target_size=(16,16))

The Dataset Class

Simple Dataset

Python 
x = np.zeros((100,10,2,2))
dataset1 = tf.data.Dataset.from_tensor_slices(x)

x2 = [np.zeros((10,2,2)), np.zeros((5,2,2))]
dataset2 = tf.data.Dataset.from_tensor_slices(x2)

Zipped Dataset

Python 
dataset_zipped = tf.data.Dataset.zip((dataset1, dataset2))

Create a Dataset from Numpy Arrays

Python 
(train_features, train_labels), (test_features, test_labels) = tf.keras.datasets.mnist.load_data()

mnist_dataset = tf.data.Dataset.from_tensor_slices((train_features, train_labels))

Create a Dataset from Text Data

Python 
text_files = sorted([f.path for f in os.scandir('data/shakespeare')])

with open(text_files[0], 'r') as fil:
    contents = [fil.readline() for i in range(5)]
    for line in contents:
        print(line)

shakespeare_dataset = tf.data.TextLineDataset(text_files)

Use the take method to get and print the first 5 lines of the dataset:

Python 
first_5_lines_dataset = iter(shakespeare_dataset.take(5))
lines = [line for line in first_5_lines_dataset]
for line in lines:
    print(line)

Interleave Lines from the Text Data Files

Python 
text_files_dataset = tf.data.Dataset.from_tensor_slices(text_files)
files = [file for file in text_files_dataset]
for file in files:
    print(file)

Interleave the lines from the text files:

Python 
interleaved_shakespeare_dataset=text_files_dataset.interleave(tf.data.TextLineDataset, cycle_length=9)

Training with Datasets

Load the UCI Bank Marketing Dataset

Python 
bank_dataframe = pd.read_csv('data/bank/bank-full.csv', delimiter=';')

Select features from the DataFrame:

Python 
features = ['age', 'job', 'marital', 'education', 'default', 'balance', 'housing',
            'loan', 'contact', 'campaign', 'pdays', 'poutcome']
labels = ['y']

bank_dataframe = bank_dataframe.filter(features + labels)

Preprocess the Data

Convert the categorical features in the DataFrame to one-hot encodings:

Python 
encoder = LabelBinarizer()
categorical_features = ['default', 'housing', 'job', 'loan', 'education', 'contact', 'poutcome']

for feature in categorical_features:
    bank_dataframe[feature] = tuple(encoder.fit_transform(bank_dataframe[feature]))

Shuffle the DataFrame:

Python 
bank_dataframe = bank_dataframe.sample(frac=1).reset_index(drop=True)

Create the Dataset Object

Convert the DataFrame to a Dataset:

Python 
bank_dataset = tf.data.Dataset.from_tensor_slices(dict(bank_dataframe))

Filter the Dataset to retain only entries with a ‘divorced’ marital status:

Python 
bank_dataset = bank_dataset.filter(lambda x : tf.equal(x['marital'], tf.constant([b'divorced']))[0] )

Convert the label (‘y’) to an integer instead of ‘yes’ or ‘no’:

Python 
def map_label(x):
    x['y'] = 0 if (x['y'] == tf.constant([b'no'], dtype= tf.string)) else 1
    return x

bank_dataset = bank_dataset.map(map_label)

Remove the ‘marital’ column:

Python 
bank_dataset = bank_dataset.map(lambda x: {key:val for key,val in x.items() if key!= 'marital'})

Create Input and Output Data Tuples

Python 
def map_feature_label(x):
    features = [[x['age']], [x['balance']], [x['campaign']], x['contact'], x['default'],
                x['education'], x['housing'], x['job'], x['loan'], [x['pdays']], x['poutcome']]
    return (tf.concat(features, axis=0), x['y'])

bank_dataset = bank_dataset.map(map_feature_label)

Split into a Training and a Validation Set

Python 
dataset_length = 0
for _ in bank_dataset:
    dataset_length += 1
print(dataset_length)

training_elements = int(dataset_length * 0.7)

train_dataset = bank_dataset.take(training_elements)
validation_dataset = bank_dataset.skip(training_elements)

Build a Classification Model

Python 
model = Sequential()
model.add(Input(shape=(30,)))
model.add(BatchNormalization(momentum=0.8))
model.add(Dense(400, activation='relu'))
model.add(BatchNormalization(momentum=0.8))
model.add(Dense(400, activation='relu'))
model.add(BatchNormalization(momentum=0.8))
model.add(Dense(1, activation='sigmoid'))

Compile the Model

Python 
optimizer = tf.keras.optimizers.Adam(1e-4)
model.compile(optimizer=optimizer, loss='binary_crossentropy', metrics=['accuracy'])

Fit the Model

Create batched training and validation datasets:

Python 
train_dataset = train_dataset.batch(20, drop_remainder=True)
validation_dataset = validation_dataset.batch(100)

Shuffle the training data:

Python 
train_dataset = train_dataset.shuffle(1000)

Fit the model:

Python 
history = model.fit(train_dataset, validation_data = validation_dataset, epochs=5)

Leave a Reply

Your email address will not be published. Required fields are marked *

Natville

© Natan Asrat 2026, Design By KitPapa