0

私は kafka -storm-cassandra を実行しようとしています。私の場合、tail2kafka 自体がプロデューサーであり、cosumer を開始してトピックを消費すると、以下のエラーがスローされます。私を助けてください。

ありがとう

[2015-05-13 15:28:51,784] エラーのため、/127.0.0.1 のソケットを閉じるエラー (kafka.network.Processor) java.lang.OutOfMemoryError: kafka.api.ProducerRequest$$anonfun$1$ の Java ヒープ領域$anonfun$apply$1.apply(ProducerRequest.scala:45) で kafka.api.ProducerRequest$$anonfun$1$$anonfun$apply$1.apply(ProducerRequest.scala:42) で scala.collection.TraversableLike$$anonfun$map scala.collection.immutable.Range$ByOne$class.foreach(Range.scala: 282) scala.collection.immutable.Range$$anon$1.foreach(Range.scala:27​​4) で scala.collection.TraversableLike$class.map(TraversableLike.scala:206) で scala.collection.immutable.Range.map (Range.scala:39) kafka.api で。ProducerRequest$$anonfun$1.apply(ProducerRequest.scala:42) で kafka.api.ProducerRequest$$anonfun$1.apply(ProducerRequest.scala:38) で scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike. scala:227) で scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:227) で scala.collection.immutable.Range$ByOne$class.foreach(Range.scala:282) で scala.collection .immutable.Range$$anon$1.foreach(Range.scala:27​​4) で scala.collection.TraversableLike$class.flatMap(TraversableLike.scala:227) で scala.collection.immutable.Range.flatMap(Range.scala:39) ) kafka.api.ProducerRequest$.readFrom(ProducerRequest.scala:38) で kafka.api.RequestKeys$$anonfun$1.apply(RequestKeys.scala:36) で kafka.api.RequestKeys$$anonfun$1.apply(RequestKeys .scala:36) kafka.network.RequestChannel$Request.(RequestChannel.scala:53) で kafka.network.Processor.read(SocketServer.scala:353) で kafka.network.Processor.run(SocketServer.scala:245) で Java で.lang.Thread.run(Thread.java:745)

そして、私の消費者コードは

インポート構造体のインポート時間

import kafka.io import kafka.request_type

クラス コンシューマー (kafka.io.IO):

CONSUME_REQUEST_TYPE = kafka.request_type.FETCH

最大サイズ = 1024 * 1024

# 秒。DEFAULT_POLLING_INTERVAL = 2

def init (self, topic, partition=0, host='localhost', port=9092): kafka.io.IO. init (自己、ホスト、ポート)

#: The topic queue to consume.
self.topic        = topic

#: The partition the topic queue is on.
self.partition    = partition

#: Offset in the Kafka queue in bytes?
self.offset       = 1

#: Maximum message size to consume.
self.max_size     = self.MAX_SIZE
self.request_type = self.CONSUME_REQUEST_TYPE
self.polling      = self.DEFAULT_POLLING_INTERVAL

self.connect()

def consumer(self): """ トピック キューからデータを消費します。 """

self.send_consume_request()

return self.parse_message_set_from(self.read_data_response())

def loop(self): """ キューからの着信メッセージをブロッキング方式でループします。pollingチェック間隔を秒単位で設定します。 """

while True:
  messages = self.consume()

  if messages and isinstance(messages, list) and len(messages) > 0:
    for message in messages:
      yield message

  time.sleep(self.polling)

# リクエスト タイプ ID + トピックの長さ + トピック + パーティション + オフセット + 最大サイズ def request_size(self): return 2 + 2 + len(self.topic) + 4 + 8 + 4

def encode_request_size(self): return struct.pack('>i', self.request_size())

def encode_request (self): 長さ = len (self.topic)

return struct.pack('>HH%dsiQi' % length, self.request_type, length, self.topic, self.partition, self.offset, self.max_size)

def send_consume_request(self): self.write(self.encode_request_size()) self.write(self.encode_request())

def read_data_response(self): buf_length = struct.unpack('>i', self.read(4))[0]

# Start with a 2 byte offset
return self.read(buf_length)[2:]

def parse_message_set_from(self, data): メッセージ = [] 処理済み = 0 長さ = len(データ) - 4

while (processed <= length):
  message_size = struct.unpack('>i', data[processed:processed+4])[0]
  messages.append(kafka.message.parse_from(data[processed:processed + message_size + 4]))
  processed += 4 + message_size

self.offset += processed

return messages
4

0 に答える 0