1. Triển khai hàng đợi

Hệ thống hàng đợi trong TensorFlow hỗ trợ quản lý luồng dữ liệu hiệu quả:

import tensorflow as tf

# Tạo hàng đợi FIFO với dung lượng tối đa 2 phần tử kiểu integer
queue = tf.FIFOQueue(2, 'int32')

# Đưa giá trị [0, 10] vào hàng đợi
initialize_op = queue.enqueue_many(([0, 10],))

# Lấy phần tử ra khỏi hàng đợi và tăng giá trị lên 1
dequeued_value = queue.dequeue()
incremented_value = dequeued_value + 1
requeue_op = queue.enqueue([incremented_value])

with tf.Session() as session:
    initialize_op.run()
    for _ in range(5):
        result, _ = session.run([dequeued_value, requeue_op])
        print(result)

2. Đồng bộ đa luồng với Coordinator và QueueRunner

Lớp tf.Coordinator giúp quản lý việc dừng đồng thời nhiều luồng xử lý:

import tensorflow as tf
import numpy as np
import threading
import time

def worker_thread(coordinator, thread_id):
    while not coordinator.should_stop():
        if np.random.rand() < 0.1:
            print(f'Dừng từ luồng: {thread_id}')
            coordinator.request_stop()
        else:
            print(f'Đang xử lý tại luồng: {thread_id}')
        time.sleep(1)

coordinator = tf.train.Coordinator()
threads = [threading.Thread(target=worker_thread, args=(coordinator, i)) for i in range(5)]

for thread in threads:
    thread.start()

coordinator.join(threads)

QueueRunner cho phép khởi chạy nhiều luồng thao tác trên cùng một hàng đợi:

import tensorflow as tf

data_queue = tf.FIFOQueue(100, 'float')
enqueue_operation = data_queue.enqueue([tf.random_normal([1])])

# Khởi tạo QueueRunner với 5 luồng thực hiện enqueue_operation
queue_runner = tf.train.QueueRunner(data_queue, [enqueue_operation] * 5)
tf.train.add_queue_runner(queue_runner)

dequeue_operation = data_queue.dequeue()

with tf.Session() as session:
    coordinator = tf.train.Coordinator()
    runner_threads = tf.train.start_queue_runners(sess=session, coord=coordinator)
    
    for _ in range(3):
        print(session.run(dequeue_operation)[0])
    
    coordinator.request_stop()
    coordinator.join(runner_threads)

3. Ghi dữ liệu định dạng TFRecord

Quy trình lưu trữ dữ liệu huấn luyện dưới dạng TFRecord:

import tensorflow as tf

def int64_feature(value):
    return tf.train.Feature(int64_list=tf.train.Int64List(value=[value]))

number_of_files = 2
samples_per_file = 2

for file_index in range(number_of_files):
    file_path = f'/đường/dẫn/data.tfrecords-{file_index:05d}-of-{number_of_files:05d}'
    writer = tf.python_io.TFRecordWriter(file_path)
    
    for sample_index in range(samples_per_file):
        record = tf.train.Example(features=tf.train.Features(
            feature={
                'file_id': int64_feature(file_index),
                'sample_id': int64_feature(sample_index)
            }
        ))
        writer.write(record.SerializeToString())
    
    writer.close()

4. Đọc và xử lý file TFRecord

Sử dụng các hàm tiện ích để đọc dữ liệu từ file TFRecord:

import tensorflow as tf

file_patterns = tf.train.match_filenames_once('/đường/dẫn/data.tfrecords-*')
input_queue = tf.train.string_input_producer(file_patterns, shuffle=False)

record_reader = tf.TFRecordReader()
_, serialized_data = record_reader.read(input_queue)

parsed_features = tf.parse_single_example(
    serialized_data,
    features={
        'file_id': tf.FixedLenFeature([], tf.int64),
        'sample_id': tf.FixedLenFeature([], tf.int64)
    }
)

with tf.Session() as session:
    tf.global_variables_initializer().run()
    coordinator = tf.train.Coordinator()
    queue_threads = tf.train.start_queue_runners(sess=session, coord=coordinator)
    
    for _ in range(6):
        print(session.run([parsed_features['file_id'], parsed_features['sample_id']]))
    
    coordinator.request_stop()
    coordinator.join(queue_threads)

5. Tổ chức dữ liệu thành batch

Tạo batch dữ liệu để cải thiện hiệu suất huấn luyện:

import tensorflow as tf

sample_data, label_data = parsed_features['file_id'], parsed_features['sample_id']
batch_dimension = 3
queue_capacity = 1000 + 3 * batch_dimension

# Tạo batch không xáo trộn
sample_batch, label_batch = tf.train.batch(
    [sample_data, label_data],
    batch_size=batch_dimension,
    capacity=queue_capacity
)

with tf.Session() as session:
    tf.global_variables_initializer().run()
    coordinator = tf.train.Coordinator()
    threads = tf.train.start_queue_runners(sess=session, coord=coordinator)
    
    for _ in range(2):
        batch_samples, batch_labels = session.run([sample_batch, label_batch])
        print(batch_samples, batch_labels)
    
    coordinator.request_stop()
    coordinator.join(threads)

Sử dụng batch có xáo trộn dữ liệu:

sample_data, label_data = parsed_features['file_id'], parsed_features['sample_id']
minimum_elements = 30
queue_capacity = minimum_elements + 3 * batch_dimension

# Tạo batch có xáo trộn
shuffled_sample_batch, shuffled_label_batch = tf.train.shuffle_batch(
    [sample_data, label_data],
    batch_size=batch_dimension,
    capacity=queue_capacity,
    min_after_dequeue=minimum_elements
)

6. Framework xử lý dữ liệu đầu vào toàn diện

Quy trình xử lý dữ liệu đầu vào hoàn chỉnh:

import tensorflow as tf

# Lấy danh sách file dữ liệu
data_files = tf.train.match_filenames_once('/đường/dẫn/file_pattern-*')
file_queue = tf.train.string_input_producer(data_files, shuffle=False)

# Đọc và phân tích dữ liệu
reader = tf.TFRecordReader()
_, raw_sample = reader.read(file_queue)

parsed_data = tf.parse_single_example(
    raw_sample,
    features={
        'image': tf.FixedLenFeature([], tf.string),
        'label': tf.FixedLenFeature([], tf.int64),
        'height': tf.FixedLenFeature([], tf.int64),
        'width': tf.FixedLenFeature([], tf.int64),
        'channels': tf.FixedLenFeature([], tf.int64)
    }
)

# Xử lý ảnh
image_content, image_label = parsed_data['image'], parsed_data['label']
img_height, img_width = parsed_data['height'], parsed_data['width']
img_channels = parsed_data['channels']

decoded_img = tf.decode_raw(image_content, tf.uint8)
decoded_img.set_shape([img_height, img_width, img_channels])

processed_image = preprocess_for_train(decoded_img, 299, 299, None)

# Tạo batch cho mạng neural
min_queue_size = 10000
training_batch_size = 100
batch_capacity = min_queue_size + 3 * training_batch_size

image_batches, label_batches = tf.train.shuffle_batch(
    [processed_image, image_label],
    batch_size=training_batch_size,
    capacity=batch_capacity,
    min_after_dequeue=min_queue_size
)

# Huấn luyện mô hình
model_output = inference(image_batches)
training_loss = calc_loss(model_output, label_batches)
optimization_step = tf.train.GradientDescentOptimizer(learning_rate).minimize(training_loss)

with tf.Session() as session:
    tf.global_variables_initializer().run()
    coordinator = tf.train.Coordinator()
    processing_threads = tf.train.start_queue_runners(sess=session, coord=coordinator)
    
    for iteration in range(training_iterations):
        session.run(optimization_step)
    
    coordinator.request_stop()
    coordinator.join(processing_threads)