我正在尝试将一个项目从apachestorm迁移到twitter heron。经过一番努力,我终于摆脱了大部分的错误,比如使用 className: "org.apache.storm.kafka.ZkHosts"
而不是 className: "storm.kafka.ZkHosts"
. 但是,我在提交拓扑时遇到了麻烦。我使用flux将拓扑提交给storm。
当它在zkstate中创建一个curatorframework对象时,我得到了一个nullpointerexception。在进一步的挖掘中,我在github中发现了一个问题,它说,这个问题是由于没有设置zookeeper的配置造成的。
进一步调试我发现这个问题是因为我缺少zkstate中需要的以下配置。java:46.
storm.zookeeper.session.timeout
storm.zookeeper.connection.timeout
storm.zookeeper.retry.times
storm.zookeeper.retry.interval
虽然我已经设法确定了这个问题,但是,我不确定在我的配置中添加这个。有人能帮我在哪里添加上面的配置吗。谢谢您。
我的流量配置
name: "My_Topology"
components:
- id: "zkHosts"
className: "org.apache.storm.kafka.ZkHosts"
constructorArgs:
- "localhost:2181"
- id: "SpoutConfig"
className: "org.apache.storm.kafka.SpoutConfig"
constructorArgs:
- ref: "zkHosts" # brokerHosts
- "my-topic" # topic
- "/my-zkRoot" # zkRoot
- "my-id" # spoutId
properties:
- name: "zkServers"
value: ["localhost"]
- name: "zkPort"
value: 2181
- name: "zkRoot"
value: "/my-zkRoot"
- name: "retryInitialDelayMs"
value: 2000
- name: "retryDelayMultiplier"
value: 2
config:
topology.workers: 5
topology.testing.always.try.serialize: true
spouts:
- id: "kafka-spout"
className: "org.apache.storm.kafka.KafkaSpout"
parallelism: 1
constructorArgs:
- ref: "SpoutConfig"
bolts:
- id: "my-bolt"
className: "com.example.sample.MyBolt"
parallelism: 1
streams:
- name: "kafka_spout --> my_bolt"
from: "kafka-spout"
to: "my-bolt"
grouping:
type: SHUFFLE
2条答案
按热度按时间nsc4cvqm1#
您可以将它们添加到fluxyaml文件的config部分
4c8rllxm2#
我们已经将风暴通量整合到heron中,使其变得简单。使用heron eco,您可以为两个不同的api编写拓扑
原生heron api
风暴api
有关更多信息,请查看此处的eco文档
https://apache.github.io/incubator-heron/docs/developers/java/eco-api/