Data preprocessing is a crucial step in the deep learning pipeline. TensorFlow provides powerful tools for data preprocessing and loading.
Data Loading Methods
1. Loading from NumPy Arrays
import numpy as np import tensorflow as tf # Create NumPy arrays x_train = np.random.rand(1000, 28, 28, 1).astype(np.float32) y_train = np.random.randint(0, 10, size=(1000,)) # Create Dataset dataset = tf.data.Dataset.from_tensor_slices((x_train, y_train)) # Print data shapes for x, y in dataset.take(1): print("X shape:", x.shape) print("Y shape:", y.shape)
2. Loading from Files
Loading from CSV Files
# Create Dataset from CSV file csv_dataset = tf.data.experimental.make_csv_dataset( 'data.csv', batch_size=32, label_name='label', num_epochs=1, ignore_errors=True ) # Or use TextLineDataset def parse_csv(line): # Parse CSV line parsed_line = tf.io.decode_csv(line, record_defaults=[0.0, 0.0, 0.0, 0]) features = parsed_line[:-1] label = parsed_line[-1] return features, label csv_dataset = tf.data.TextLineDataset('data.csv').skip(1).map(parse_csv)
Loading from Image Files
# Create Dataset from image files image_paths = tf.data.Dataset.list_files('images/*.jpg') def load_image(path): # Read image image = tf.io.read_file(path) # Decode image image = tf.image.decode_jpeg(image, channels=3) # Resize image = tf.image.resize(image, [224, 224]) # Normalize image = image / 255.0 return image image_dataset = image_paths.map(load_image)
Loading from TFRecord Files
# Create Dataset from TFRecord file tfrecord_dataset = tf.data.TFRecordDataset('data.tfrecord') def parse_tfrecord(example_proto): # Define feature parsing feature_description = { 'image': tf.io.FixedLenFeature([], tf.string), 'label': tf.io.FixedLenFeature([], tf.int64), } # Parse example example = tf.io.parse_single_example(example_proto, feature_description) # Decode image image = tf.io.decode_jpeg(example['image'], channels=3) image = tf.image.resize(image, [224, 224]) image = image / 255.0 return image, example['label'] tfrecord_dataset = tfrecord_dataset.map(parse_tfrecord)
3. Loading from Pandas DataFrame
import pandas as pd # Create DataFrame df = pd.DataFrame({ 'feature1': np.random.rand(1000), 'feature2': np.random.rand(1000), 'label': np.random.randint(0, 2, size=1000) }) # Create Dataset from DataFrame dataset = tf.data.Dataset.from_tensor_slices(( df[['feature1', 'feature2']].values, df['label'].values ))
Data Preprocessing Methods
1. Image Preprocessing
# Image data augmentation def augment_image(image, label): # Random flip image = tf.image.random_flip_left_right(image) image = tf.image.random_flip_up_down(image) # Random rotation image = tf.image.rot90(image, k=tf.random.uniform(shape=[], minval=0, maxval=4, dtype=tf.int32)) # Random brightness adjustment image = tf.image.random_brightness(image, max_delta=0.2) # Random contrast adjustment image = tf.image.random_contrast(image, lower=0.8, upper=1.2) # Random saturation adjustment image = tf.image.random_saturation(image, lower=0.8, upper=1.2) # Random crop image = tf.image.random_crop(image, size=[200, 200, 3]) image = tf.image.resize(image, [224, 224]) return image, label # Apply data augmentation augmented_dataset = dataset.map(augment_image)
2. Text Preprocessing
# Text preprocessing import tensorflow_text as text # Text normalization def normalize_text(text): # Convert to lowercase text = tf.strings.lower(text) # Remove punctuation text = tf.strings.regex_replace(text, r'[^\w\s]', '') # Remove extra spaces text = tf.strings.strip(text) return text # Text tokenization def tokenize_text(text): # Use Unicode tokenizer tokenizer = text.UnicodeScriptTokenizer() tokens = tokenizer.tokenize(text) return tokens # Build vocabulary def build_vocabulary(dataset, vocab_size=10000): # Count word frequency vocab = collections.Counter() for text in dataset: tokens = tokenize_text(normalize_text(text)) vocab.update(tokens.numpy()) # Select most common words most_common = vocab.most_common(vocab_size) vocab_list = [word for word, _ in most_common] # Add special tokens vocab_list = ['<PAD>', '<UNK>', '<START>', '<END>'] + vocab_list return vocab_list # Text encoding def encode_text(text, vocab, max_length=100): # Tokenize tokens = tokenize_text(normalize_text(text)) # Convert to indices indices = [vocab.get(token, vocab['<UNK>']) for token in tokens] # Truncate or pad if len(indices) > max_length: indices = indices[:max_length] else: indices = indices + [vocab['<PAD>']] * (max_length - len(indices)) return tf.constant(indices)
3. Numerical Data Preprocessing
# Numerical data standardization def normalize_features(features): # Calculate mean and standard deviation mean = tf.reduce_mean(features, axis=0) std = tf.math.reduce_std(features, axis=0) # Standardize normalized = (features - mean) / (std + 1e-7) return normalized # Numerical data normalization def min_max_normalize(features): # Calculate min and max min_val = tf.reduce_min(features, axis=0) max_val = tf.reduce_max(features, axis=0) # Normalize to [0, 1] normalized = (features - min_val) / (max_val - min_val + 1e-7) return normalized # Numerical data standardization (using precomputed statistics) class StandardScaler: def __init__(self): self.mean = None self.std = None def fit(self, data): self.mean = tf.reduce_mean(data, axis=0) self.std = tf.math.reduce_std(data, axis=0) def transform(self, data): return (data - self.mean) / (self.std + 1e-7) def fit_transform(self, data): self.fit(data) return self.transform(data)
4. Category Encoding
# One-Hot encoding def one_hot_encode(labels, num_classes): return tf.one_hot(labels, num_classes) # Label encoding def label_encode(labels, label_map): return tf.map_fn(lambda x: label_map[x.numpy()], labels, dtype=tf.int32) # Build label mapping def build_label_map(labels): unique_labels = tf.unique(labels).y label_map = {label: idx for idx, label in enumerate(unique_labels.numpy())} return label_map
Dataset Operations
1. Batching
# Batching batched_dataset = dataset.batch(32) # Padded batching (for variable-length sequences) padded_batch_dataset = dataset.padded_batch( batch_size=32, padded_shapes=([None], []), # Padding shapes for features and labels padding_values=(0.0, 0) # Padding values )
2. Shuffling
# Shuffle data shuffled_dataset = dataset.shuffle(buffer_size=1000) # Shuffle and batch shuffled_batched_dataset = dataset.shuffle(buffer_size=1000).batch(32)
3. Repeating
# Repeat data repeated_dataset = dataset.repeat(count=2) # Repeat 2 times # Infinite repeat infinite_dataset = dataset.repeat()
4. Mapping
# Apply function to each element mapped_dataset = dataset.map(lambda x, y: (x * 2, y)) # Parallel mapping parallel_mapped_dataset = dataset.map( lambda x, y: (x * 2, y), num_parallel_calls=tf.data.AUTOTUNE )
5. Filtering
# Filter data filtered_dataset = dataset.filter(lambda x, y: y > 5) # Filter and map filtered_mapped_dataset = dataset.filter( lambda x, y: y > 5 ).map(lambda x, y: (x, y - 5))
6. Taking and Skipping
# Take first N elements taken_dataset = dataset.take(100) # Skip first N elements skipped_dataset = dataset.skip(100) # Take first N and skip first M taken_skipped_dataset = dataset.skip(100).take(50)
7. Prefetching
# Prefetch data (improve performance) prefetched_dataset = dataset.prefetch(tf.data.AUTOTUNE) # Complete data pipeline optimized_dataset = ( dataset .shuffle(buffer_size=1000) .map(preprocess, num_parallel_calls=tf.data.AUTOTUNE) .batch(32) .prefetch(tf.data.AUTOTUNE) )
Efficient Data Loading Tips
1. Using Cache
# Cache data (suitable for small datasets) cached_dataset = dataset.cache() # Cache to file file_cached_dataset = dataset.cache('cache_dir')
2. Parallel Processing
# Parallel mapping parallel_dataset = dataset.map( preprocess, num_parallel_calls=tf.data.AUTOTUNE ) # Parallel reading parallel_read_dataset = tf.data.Dataset.list_files( 'images/*.jpg', shuffle=False ).interleave( tf.data.TFRecordDataset, cycle_length=4, num_parallel_calls=tf.data.AUTOTUNE )
3. Data Compression
# Compress data (reduce I/O) compressed_dataset = dataset.interleave( tf.data.TFRecordDataset, cycle_length=4, num_parallel_calls=tf.data.AUTOTUNE )
4. Using Generators
# Create Dataset from Python generator def data_generator(): for i in range(1000): yield np.random.rand(28, 28, 1), np.random.randint(0, 10) generator_dataset = tf.data.Dataset.from_generator( data_generator, output_signature=( tf.TensorSpec(shape=(28, 28, 1), dtype=tf.float32), tf.TensorSpec(shape=(), dtype=tf.int64) ) )
Complete Data Preprocessing Pipeline
import tensorflow as tf import numpy as np # 1. Load data def load_data(): # Create simulated data x_train = np.random.rand(1000, 28, 28, 1).astype(np.float32) y_train = np.random.randint(0, 10, size=(1000,)) # Create Dataset dataset = tf.data.Dataset.from_tensor_slices((x_train, y_train)) return dataset # 2. Data preprocessing def preprocess(image, label): # Normalize image = image / 255.0 # Data augmentation (only during training) if tf.random.uniform(()) > 0.5: image = tf.image.random_flip_left_right(image) image = tf.image.random_brightness(image, max_delta=0.1) return image, label # 3. Create data pipeline def create_dataset(dataset, batch_size=32, shuffle=True, augment=True): # Shuffle data if shuffle: dataset = dataset.shuffle(buffer_size=1000) # Apply preprocessing dataset = dataset.map( preprocess, num_parallel_calls=tf.data.AUTOTUNE ) # Batch dataset = dataset.batch(batch_size) # Prefetch dataset = dataset.prefetch(tf.data.AUTOTUNE) return dataset # 4. Use data pipeline # Load data train_dataset = load_data() # Create training dataset train_dataset = create_dataset(train_dataset, batch_size=32, shuffle=True, augment=True) # Create validation dataset val_dataset = create_dataset(train_dataset.take(200), batch_size=32, shuffle=False, augment=False) # Train model model.fit(train_dataset, epochs=10, validation_data=val_dataset)
Data Preprocessing Best Practices
1. Data Pipeline Optimization
# Optimized data pipeline optimized_pipeline = ( dataset .cache() # Cache data .shuffle(buffer_size=10000) # Shuffle data .map(preprocess, num_parallel_calls=tf.data.AUTOTUNE) # Parallel preprocessing .batch(32) # Batch .prefetch(tf.data.AUTOTUNE) # Prefetch data )
2. Memory Management
# Use generators to reduce memory usage def lazy_load_data(): for file_path in file_paths: data = load_file(file_path) yield data lazy_dataset = tf.data.Dataset.from_generator( lazy_load_data, output_signature=... )
3. Data Validation
# Validate data def validate_data(dataset): for x, y in dataset.take(1): print(f"X shape: {x.shape}, dtype: {x.dtype}") print(f"Y shape: {y.shape}, dtype: {y.dtype}") # Check value range print(f"X range: [{tf.reduce_min(x):.2f}, {tf.reduce_max(x):.2f}]") # Check for NaN or Inf if tf.reduce_any(tf.math.is_nan(x)): print("Warning: NaN detected in X!") if tf.reduce_any(tf.math.is_inf(x)): print("Warning: Inf detected in X!") # Use validation validate_data(train_dataset)
4. Data Visualization
import matplotlib.pyplot as plt # Visualize data def visualize_data(dataset, num_samples=5): fig, axes = plt.subplots(1, num_samples, figsize=(15, 3)) for i, (x, y) in enumerate(dataset.take(num_samples)): axes[i].imshow(x.numpy().squeeze(), cmap='gray') axes[i].set_title(f'Label: {y.numpy()}') axes[i].axis('off') plt.tight_layout() plt.show() # Use visualization visualize_data(train_dataset)
Summary
TensorFlow provides powerful data preprocessing and loading tools:
- Data Loading: Support for multiple data sources (NumPy, files, TFRecord, etc.)
- Data Preprocessing: Preprocessing methods for images, text, and numerical data
- Dataset Operations: Batching, shuffling, mapping, filtering, etc.
- Efficient Loading: Caching, parallel processing, prefetching, etc.
- Best Practices: Data pipeline optimization, memory management, data validation
Mastering these data preprocessing techniques will help you build and train deep learning models more efficiently.