Kafka消费者无法解压读取.gz文件，报“IOError: Not a gzipped file”错误

嘿，这个问题我之前也碰到过！你看到的IOError: Not a gzipped file错误，核心原因是kafka-console-producer.sh默认会把输入当成文本内容逐行读取并发送，而不是直接传输.gz文件的原始二进制字节流。

当你用< ~/Downloads/stocks.json.gz重定向输入时，控制台生产者会把压缩文件的二进制数据当成文本，按换行符拆分消息，甚至可能做了字符编码转换——这就彻底破坏了gzip文件的结构，消费者收到的自然不是合法的gzip数据，解压失败也就不足为奇了。

解决方案1：用控制台生产者发送二进制数据 #

如果你还是想用kafka-console-producer.sh来发送，只需要指定用二进制序列化器处理消息值，确保文件的原始二进制数据被完整发送：

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic Airport \
  --property parse.key=false \
  --property value.serializer=org.apache.kafka.common.serialization.ByteArraySerializer < ~/Downloads/stocks.json.gz

这个配置告诉生产者直接把输入的字节流作为消息值发送，不会做任何文本解析或拆分，消费者就能拿到完整的gzip文件了。

解决方案2：用Python生产者发送压缩文件 #

如果控制台方式不好用，也可以写个简单的Python生产者，直接读取.gz文件的二进制内容发送：

from kafka import KafkaProducer

producer = KafkaProducer(bootstrap_servers='localhost:9092')
topic = 'Airport'

# 以二进制模式读取压缩文件
with open('/home/your-user/Downloads/stocks.json.gz', 'rb') as f:
    file_content = f.read()
    producer.send(topic, value=file_content)
# 确保消息发送完成
producer.flush()

这种方式能完全保证文件的二进制数据被准确传输到Kafka。

消费者代码的小优化 #

你的消费者代码已经差不多了，这里补全并做个小调整，确保拿到原始字节数据：

import sys
import gzip
import StringIO
from kafka import KafkaConsumer

KAFKA_TOPIC = 'Airport'
KAFKA_BROKERS = 'localhost:9092'

# 指定value_deserializer直接返回原始字节，避免默认的字符串解码破坏数据
consumer = KafkaConsumer(
    KAFKA_TOPIC,
    bootstrap_servers=KAFKA_BROKERS,
    value_deserializer=lambda x: x
)

try:
    for message in consumer:
        # 将消息值包装成可读取的文件对象
        f = StringIO.StringIO(message.value)
        with gzip.GzipFile(fileobj=f, mode='rb') as gzip_f:
            # 读取并处理解压后的内容
            uncompressed_content = gzip_f.read()
            print(uncompressed_content)
except Exception as e:
    print(f"处理出错：{e}", file=sys.stderr)

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