Kafka使用者解压缩gz文件流并读取

Question

Kafka producer正在发送.gz文件，但无法在消费者端解压缩和读取文件。获取错误为“IOError：不是gzipped文件”

producer-bin/kafka-console-producer.sh--broker-list localhost:9092-topic Airport<~/downloads/stocks.json.gz

消费者-

import sys 
import gzip
import StringIO
from kafka import KafkaConsumer

consumer = KafkaConsumer(KAFKA_TOPIC, bootstrap_servers=KAFKA_BROKERS)

try:
    for message in consumer:
        f = StringIO.StringIO(message.value)
        gzip_f = gzip.GzipFile(fileobj=f)
        unzipped_content = gzip_f.read()
        content = unzipped_content.decode('utf8')
        print (content)
except KeyboardInterrupt:
    sys.exit()

使用者出错-

Traceback (most recent call last):
  File "consumer.py", line 18, in <module>
    unzipped_content = gzip_f.read()
  File "/usr/lib64/python2.6/gzip.py", line 212, in read
    self._read(readsize)
  File "/usr/lib64/python2.6/gzip.py", line 255, in _read
    self._read_gzip_header()
  File "/usr/lib64/python2.6/gzip.py", line 156, in _read_gzip_header
    raise IOError, 'Not a gzipped file'
IOError: Not a gzipped file

Answer 1

Kafaka并不是用来发送巨大的有效载荷/信息的。您应该将其视为分布式消息总线，它赋予您分布式系统的所有特权。

Kafaka限制可以跨行发送的信息的大小，原因如下

• 巨大的消息会增加代理中的内存压力。
• 较大的消息会降低代理的速度，并且处理这些消息的成本非常高。

null

使用批处理大小：

batch.size以总字节数而不是消息数度量批处理大小。它控制在向Kafka代理发送消息之前要收集多少字节的数据。在不超过可用内存的情况下，将其设置得尽可能高。默认值为16384。

如果增加缓冲区的大小，它可能永远不会满。生产者最终会根据其他触发器发送信息，例如以毫秒为单位的延迟时间。虽然可以通过设置太高的缓冲区批处理大小来降低内存使用，但这不会影响延迟。

如果您的生产者一直在发送，那么您可能会获得尽可能好的吞吐量。如果生产者经常空闲，您可能没有写入足够的数据来保证当前的资源分配。

因为您的数据是gzip，所以您可以使用referenceBasedMessaging。

而不是玩提取大小和消息最大字节大小（不能覆盖所有文件大小），将您的文件存储在分布式文件系统（如NFS/HDFS/S3)上，并将引用发送给用户。用户可以选择位置并解压数据。