Сделать топологию Apache Storm, чтобы использовать последнее смещение от Кафки - PullRequest
Новая
       14

Сделать топологию Apache Storm, чтобы использовать последнее смещение от Кафки

0 голосов
/ 26 мая 2018

У меня есть kafkaspout, 2 болта для обработки данных, 2 болта для хранения обработанных данных в mongodb

Я использую Apache Flux для создания топологии, где я считываю данные в носик из kafka.Все работает нормально, но каждый раз, когда я запускаю топологию, он обрабатывает все сообщения в kafka с самого начала.и как только он обрабатывает все сообщения, он не ждет больше сообщений и сбоев.

Error Message

Как я могу сделать топологию шторма для обработкитолько последние сообщения.

вот мой файл топологии .yaml 

name: "kafka-topology"

components:
# MongoDB mapper
  - id: "block-mapper"
    className: "org.apache.storm.mongodb.common.mapper.SimpleMongoMapper"
    configMethods:
      - name: "withFields"
        args: # The following are the tuple fields to map to a MongoDB document
          - ["block"]
# MongoDB mapper
  - id: "transaction-mapper"
    className: "org.apache.storm.mongodb.common.mapper.SimpleMongoMapper"
    configMethods:
      - name: "withFields"
        args: # The following are the tuple fields to map to a MongoDB document
          - ["transaction"]

  - id: "stringScheme"
    className: "org.apache.storm.kafka.StringScheme"

  - id: "stringMultiScheme"
    className: "org.apache.storm.spout.SchemeAsMultiScheme"
    constructorArgs:
      - ref: "stringScheme"

  - id: "zkHosts"
    className: "org.apache.storm.kafka.ZkHosts"
    constructorArgs:
      - "172.25.33.191:2181"

  - id: "spoutConfig"
    className: "org.apache.storm.kafka.SpoutConfig"
    constructorArgs:
      # brokerHosts
      - ref: "zkHosts"
      # topic
      - "blockdata"
      # zkRoot
      - ""
      # id
      - "myId"
    properties:
      - name: "scheme"
        ref: "stringMultiScheme"
      - name: "ignoreZkOffsets"
        value: flase



config:
  topology.workers: 1
  # ...

# spout definitions
spouts:
  - id: "kafka-spout"
    className: "org.apache.storm.kafka.KafkaSpout"
    constructorArgs:
      - ref: "spoutConfig"
    parallelism: 1

# bolt definitions
bolts:
  - id: "blockprocessing-bolt"
    className: "org.apache.storm.flux.wrappers.bolts.FluxShellBolt"
    constructorArgs:
      # command line
      - ["python", "process-bolt.py"]
      # output fields
      - ["block"]
    parallelism: 1
    # ...
  - id: "transprocessing-bolt"
    className: "org.apache.storm.flux.wrappers.bolts.FluxShellBolt"
    constructorArgs:
      # command line
      - ["python", "trans-bolt.py"]
      # output fields
      - ["transaction"]
    parallelism: 1
    # ...

  - id: "mongoBlock-bolt"
    className: "org.apache.storm.mongodb.bolt.MongoInsertBolt"
    constructorArgs:
      - "mongodb://172.25.33.205:27017/testdb"
      - "block"
      - ref: "block-mapper"
    parallelism: 1
    # ...

  - id: "mongoTrans-bolt"
    className: "org.apache.storm.mongodb.bolt.MongoInsertBolt"
    constructorArgs:
      - "mongodb://172.25.33.205:27017/testdb"
      - "transaction"
      - ref: "transaction-mapper"
    parallelism: 1
    # ...



  - id: "log"
    className: "org.apache.storm.flux.wrappers.bolts.LogInfoBolt"
    parallelism: 1
    # ...

#stream definitions
# stream definitions define connections between spouts and bolts.
# note that such connections can be cyclical
# custom stream groupings are also supported

streams:


  - name: "kafka --> block-Processing" # name isn't used (placeholder for logging, UI, etc.)
    from: "kafka-spout"
    to: "blockprocessing-bolt"
    grouping:
      type: SHUFFLE

  - name: "kafka --> transaction-processing" # name isn't used (placeholder for logging, UI, etc.)
    from: "kafka-spout"
    to: "transprocessing-bolt"
    grouping:
      type: SHUFFLE

  - name: "block --> mongo"
    from: "blockprocessing-bolt"
    to: "mongoBlock-bolt"
    grouping:
      type: SHUFFLE

  - name: "transaction --> mongo"
    from: "transprocessing-bolt"
    to: "mongoTrans-bolt"
    grouping:
      type: SHUFFLE

Я попытался добавить свойство в spoutconfig для получения только самых последних сообщений, как это 

 - id: "spoutConfig"
className: "org.apache.storm.kafka.SpoutConfig"
constructorArgs:
  - ref: "zkHosts"
  - "blockdata"
  - ""
  - "myId"
properties:
  - name: "scheme"
    ref: "stringMultiScheme"
  - name: "startOffsetTime"
    ref: "EarliestTime"

  - name: "forceFromStart"
    value: false

Но это выдает ошибку независимо от того, что я помещаю в ссылку startOffsetTime

Exception in thread "main" java.lang.IllegalArgumentException: Can not set long field org.apache.storm.kafka.KafkaConfig.startOffsetTime to null value

1 Ответ

0 голосов
/ 26 мая 2018

Вам необходимо установить для startOffsetTime значение kafka.api.OffsetRequest.LatestTime.Как вы можете видеть на https://github.com/apache/storm/tree/64af629a19a82591dbf3428f7fd6b02f39e0723f/external/storm-kafka#kafkaconfig,, настройка по умолчанию перейдет на самое раннее из доступных смещений.

Исключение, которое вы нажимаете, кажется, не связано.Похоже на несовместимость куратора / Zookeeper.

Редактировать: Я думаю, что вы решаете эту проблему https://issues.apache.org/jira/browse/STORM-2978. 1.2.2 скоро выйдет, попробуйте обновить, как только он выйдет.

Редактировать редактирование: Если вы хотитеобходите его без обновления, отредактируйте pom для своей топологии, чтобы он включал зависимость от Zookeeper 3.4, а не 3.5.

...