在TensorFlow中读取LZO压缩CSV文件遇阻，寻求解决方案

你遇到的问题很典型：TensorFlow默认的文本读取/CSV解析逻辑只能处理原始文本流，而LZO是二进制压缩格式，必须先完成解压步骤才能正常解析。下面给你两种实用的解决方案，按需选择：

方法一：提前解压文件（适合小批量数据） #

如果你的文件数量不多、体积不大，最直接的方式是先用Python的LZO库把文件解压成普通CSV，再用你原来的代码处理：

1. 先安装依赖库：

pip install python-lzo

2. 批量解压LZO文件的示例代码：

import lzo
import os

def decompress_lzo(input_path, output_path):
    with open(input_path, 'rb') as f_in:
        compressed_data = f_in.read()
        # 解压数据，部分特殊LZO文件可能需要指定buffer_size参数
        decompressed_data = lzo.decompress(compressed_data)
    with open(output_path, 'wb') as f_out:
        f_out.write(decompressed_data)

# 遍历目标文件夹下的所有LZO文件
for filename in os.listdir('./lzo_datasets'):
    if filename.endswith('.lzo'):
        input_file = f'./lzo_datasets/{filename}'
        output_file = f'./uncompressed_datasets/{filename[:-4]}.csv'
        decompress_lzo(input_file, output_file)

解压完成后，你原来的parse_csv函数就能正常读取这些CSV文件了。

方法二：流式解压处理（适合大规模/大体积数据） #

如果文件太大不适合提前解压，或者需要构建端到端的数据流管道，可以结合tf.data.Dataset和Python解压逻辑，通过tf.py_function把解压步骤嵌入TensorFlow的数据流中：

import lzo
import tensorflow as tf

# 先定义你的列默认值和列名（替换成你实际的配置）
DEFAULTS = [0.0, "", 0]  # 示例默认值，根据你的CSV结构调整
COLUMNS = ["feature1", "feature2", "label"]

def parse_csv(line):
    columns = tf.decode_csv(line, record_defaults=DEFAULTS, field_delim="\t", use_quote_delim=False)
    features = dict(zip(COLUMNS, columns))
    # TF2.x中tf.to_int32已废弃，改用tf.cast
    labels = tf.cast(features.pop('label'), tf.int32)
    return features, labels

def decompress_lzo_file(file_path):
    # 用Python逻辑读取并解压LZO文件
    file_path_str = file_path.numpy().decode('utf-8')
    with open(file_path_str, 'rb') as f:
        compressed_data = f.read()
        decompressed_text = lzo.decompress(compressed_data).decode('utf-8')
    # 按行分割文本，返回字符串列表
    return decompressed_text.split('\n')

def create_lzo_dataset(lzo_file_paths):
    dataset = tf.data.Dataset.from_tensor_slices(lzo_file_paths)
    # 调用Python解压函数，把每个LZO文件转换成文本行数据集
    dataset = dataset.flat_map(lambda path: tf.data.Dataset.from_tensor_slices(
        tf.py_function(decompress_lzo_file, [path], tf.string)
    ))
    # 过滤空行避免解析错误
    dataset = dataset.filter(lambda line: tf.strings.length(line) > 0)
    # 解析CSV行
    dataset = dataset.map(parse_csv)
    # 可根据需求添加shuffle、batch、prefetch等优化
    dataset = dataset.shuffle(1000).batch(32).prefetch(tf.data.AUTOTUNE)
    return dataset

# 使用示例
file_paths = ['./data_part1.csv.lzo', './data_part2.csv.lzo']
train_dataset = create_lzo_dataset(file_paths)

额外注意点 #

• 如果你的LZO是分块压缩的格式，可能需要调整lzo.decompress的参数（比如指定buffer_size），具体可以参考python-lzo的文档。
• 流式处理的性能会比提前解压稍慢，但胜在不需要额外的磁盘空间存储解压后的文件，适合大规模数据场景。

内容的提问来源于stack exchange，提问作者Gianluca Micchi