深入理解与实践：S3上大型Gzip文件头部与尾部的高效提取策略

import boto3
import zlib

def get_first_line_from_s3_gzip(bucket_name, file_name, chunk_size=1024):
    """
    从S3上的Gzip文件中提取解压后的首行内容。

    Args:
        bucket_name (str): S3桶的名称。
        file_name (str): S3文件键。
        chunk_size (int): 要下载并尝试解压的起始字节数。

    Returns:
        str: 解压后的首行内容。
    """
    s3 = boto3.client('s3')

    # 构建Range请求头，只下载文件的前chunk_size字节
    range_header = f"bytes=0-{chunk_size - 1}"
    try:
        response = s3.get_object(Bucket=bucket_name, Key=file_name, Range=range_header)
        content_compressed = response['Body'].read()
    except Exception as e:
        print(f"Error fetching object range from S3: {e}")
        return None

    # 使用zlib解压器处理Gzip格式
    decompressor = zlib.decompressobj(zlib.MAX_WBITS | 32)
    try:
        first_part_decompressed = decompressor.decompress(content_compressed)
        # 确保处理剩余数据（如果有），尽管通常头部不会有太多剩余
        first_part_decompressed += decompressor.flush() 

        # 解码并提取首行
        return first_part_decompressed.decode('utf-8').split('\n')[0]
    except zlib.error as e:
        print(f"Error decompressing header: {e}")
        return None

# 示例调用 (请替换为您的实际桶名和文件名)
# bucket = "your-s3-bucket"
# key = "your-large-file.gz"
# first_line = get_first_line_from_s3_gzip(bucket, key)
# if first_line:
#     print(f"文件首行: {first_line}")

import boto3
import zlib

def get_last_line_from_s3_gzip_failed(bucket_name, file_name, chunk_size=1024):
    """
    尝试从S3上的Gzip文件尾部提取解压后的内容（此方法会失败）。
    """
    s3 = boto3.client('s3')

    try:
        # 首先获取文件总大小
        response_head = s3.head_object(Bucket=bucket_name, Key=file_name)
        file_size = response_head['ContentLength']

        # 构建Range请求头，只下载文件的最后chunk_size字节
        range_start = max(0, file_size - chunk_size)
        range_header = f"bytes={range_start}-{file_size - 1}"
        response_tail = s3.get_object(Bucket=bucket_name, Key=file_name, Range=range_header)
        content_compressed_tail = response_tail['Body'].read()
    except Exception as e:
        print(f"Error fetching object range from S3: {e}")
        return None

    decompressor = zlib.decompressobj(zlib.MAX_WBITS | 32)
    try:
        # 尝试解压，这里会抛出zlib.error
        last_part_decompressed = decompressor.decompress(content_compressed_tail)
        print(f"解压成功（不应发生）: {last_part_decompressed.decode('utf-8')}")
    except zlib.error as e:
        print(f"Error decompressing footer: {e}") # 预期会在此处捕获错误
        print("原因：Gzip文件尾部不包含独立的有效头部信息，无法直接解压。")
    return None

# 示例调用 (请替换为您的实际桶名和文件名)
# bucket = "your-s3-bucket"
# key = "your-large-file.gz"
# get_last_line_from_s3_gzip_failed(bucket, key)

import boto3
import zlib
from collections import deque

def get_last_n_lines_streamed_from_s3_gzip(bucket_name, file_name, n_lines=10, s3_chunk_size=4096):
    """
    通过流式解压从S3上的Gzip文件中获取最后N行。
    此方法仍需处理整个Gzip流，但避免将整个解压内容加载到内存。
    """
    s3 = boto3.client('s3')

    try:
        response = s3.get_object(Bucket=bucket_name, Key=file_name)
        stream = response['Body']
    except Exception as e:
        print(f"Error fetching object from S3: {e}")
        return []

    decompressor = zlib.decompressobj(zlib.MAX_WBITS | 32)
    last_lines = deque(maxlen=n_lines) # 使用双端队列存储最后N行
    buffer = b'' # 用于累积不完整的行

    while True:
        chunk_compressed = stream.read(s3_chunk_size)
        if not chunk_compressed:
            break

        chunk_decompressed = decompressor.decompress(chunk_compressed)

        # 将解压后的数据添加到缓冲区
        buffer += chunk_decompressed

        # 按行分割缓冲区内容
        lines = buffer.split(b'\n')

        # 除了最后一行（可能不完整），其他行都已完整
        for line in lines[:-1]:
            last_lines.append(line.decode('utf-8', errors='ignore'))

        # 将不完整（或可能是完整的最后一段）的行保留在缓冲区
        buffer = lines[-1]

    # 处理文件结束时缓冲区中可能剩余的最后一段数据
    # 刷新decompressor以获取所有剩余的解压数据
    buffer += decompressor.flush()
    if buffer:
        final_lines = buffer.split(b'\n')
        for line in final_lines:
            if line: # 避免添加空行
                last_lines.append(line.decode('utf-8', errors='ignore'))

    return list(last_lines)

# 示例调用 (请替换为您的实际桶名和文件名)
# bucket = "your-s3-bucket"
# key = "your-large-file.gz"
# last_10_lines = get_last_n_lines_streamed_from_s3_gzip(bucket, key, n_lines=10)
# if last_10_lines:
#     print(f"文件最后 {len(last_10_lines)} 行:")
#     for line in last_10_lines:
#         print(line)