S3大型Gzip文件：高效提取首尾数据的挑战与策略

import boto3
import zlib

def get_first_n_bytes_from_s3_gzip(bucket_name: str, file_key: str, num_bytes_to_fetch: int = 2048) -> str:
    """
    从S3上的Gzip文件头部提取指定字节数的解压内容。

    Args:
        bucket_name (str): S3存储桶名称。
        file_key (str): S3文件键。
        num_bytes_to_fetch (int): 从文件头部获取的原始（压缩）字节数。
                                   此值应足够大，以包含Gzip头部和部分数据。

    Returns:
        str: 解压后的首部字符串。
    Raises:
        zlib.error: 如果解压失败。
        Exception: 如果获取S3对象失败。
    """
    s3 = boto3.client('s3')
    # S3 Range请求是包含性的，所以请求bytes=0-(N-1)表示获取N个字节
    range_header = f"bytes=0-{num_bytes_to_fetch - 1}"

    try:
        response = s3.get_object(Bucket=bucket_name, Key=file_key, Range=range_header)
        compressed_content = response['Body'].read()

        # 使用zlib.MAX_WBITS | 32来自动检测gzip头部并进行解压
        # zlib.MAX_WBITS | 32 标志告诉zlib处理gzip格式
        decompressor = zlib.decompressobj(zlib.MAX_WBITS | 32)
        decompressed_data = decompressor.decompress(compressed_content)

        # 如果还有剩余数据（通常在chunk_size不是精确块边界时），确保全部解压
        decompressed_data += decompressor.flush()

        return decompressed_data.decode('utf-8')
    except zlib.error as e:
        print(f"解压首部数据时发生错误: {e}")
        raise
    except Exception as e:
        print(f"获取S3对象时发生错误: {e}")
        raise

# 示例用法 (请替换为您的实际bucket和key)
# bucket = "your-s3-bucket"
# key = "path/to/your/large.gz"
# try:
#     header_content = get_first_n_bytes_from_s3_gzip(bucket, key, 4096) # 尝试获取4KB原始数据
#     print("文件首部内容:")
#     # 打印解压后的第一行
#     print(header_content.split('\n')[0]) 
# except Exception as e:
#     print(f"处理失败: {e}")

import s3fs
import gzip

def get_first_and_last_lines_from_s3_gzip_stream(s3_path: str, header_chunk_size: int = 1024, footer_chunk_size: int = 128) -> tuple[str, str]:
    """
    通过流式处理S3上的Gzip文件来获取首部和尾部内容。
    注意：此方法会流式传输并解压整个文件以获取尾部数据，但不会将其完全下载到本地磁盘。

    Args:
        s3_path (str): S3文件路径，例如 "s3://your-bucket/path/to/file.gz"。
        header_chunk_size (int): 获取首部时读取的解压字节数。
        footer_chunk_size (int): 获取尾部时读取的解压字节数。

    Returns:
        tuple[str, str]: (首部内容字符串, 尾部内容字符串)。
    """
    fs = s3fs.S3FileSystem()
    with fs.open(s3_path, 'rb') as s3_file:  # 'rb' 表示二进制读取模式
        # gzip.open() 可以接受文件类对象，它会在内部处理Gzip解压
        with gzip.open(s3_file, 'rt', encoding='utf-8') as f_gzip: # 'rt' 表示文本读取模式，自动处理解压和编码
            # 获取首部
            f_gzip.seek(0, 0) # 确保从文件开始位置读取
            header_raw = f_gzip.read(header_chunk_size)

            # 获取尾部
            # 注意：f_gzip.seek(-offset, 2) 会触发内部从头开始的解压过程，
            # 直到定位到文件末尾的指定偏移量。
            f_gzip.seek(-footer_chunk_size, 2) 
            footer_raw = f_gzip.read(footer_chunk_size)

            # 返回第一行和最后一行
            first_line = header_raw.split('\n')[0] if header_raw else ""
            last_line = footer_raw.split('\n')[-1] if footer_raw else ""

            return first_line, last_line

# 示例用法 (请替换为您的实际S3文件路径)
# s3_file_path = "s3://your-s3-bucket/path/to/your/large.gz"
# try:
#     first_line, last_line = get_first_and_last_lines_from_s3_gzip_stream(s3_file_path)
#     print(f"文件第一行: {first_line}")
#     print(f"文件最后一行: {last_line}")
# except Exception as e:
#     print(f"处理S3 Gzip文件失败: {e}")