企业网站重要性,关键词优化工具,企业品牌策划设计,开发软件多少钱微型框架中的助焊剂可以帮助我们定义和部署Storm拓扑。 Flux有各种包装器#xff0c;可帮助您定义所需的流并初始化Bolts和Spouts#xff08;使用带有或不带有参数的构造函数#xff0c;并通过反射自动调用自定义配置方法#xff09;。 您只需要使用Flux就是将其作为依赖… 微型框架中的助焊剂可以帮助我们定义和部署Storm拓扑。 Flux有各种包装器可帮助您定义所需的流并初始化Bolts和Spouts使用带有或不带有参数的构造函数并通过反射自动调用自定义配置方法。 您只需要使用Flux就是将其作为依赖项添加到“ pom.xml”中通过单个YAML文件进行配置请检查助焊剂示例 然后将其用作主类以在Storm集群中部署拓扑或作为本地测试。 为了初始化KafkaBolt 需要执行以下步骤 通过“ withTopicSelector ”方法定义“ topicSelector ” 通过“ withTupleToKafkaMapper ”方法定义一个“ kafkaMapper” 通过“ withProducerProperties ”方法定义一个“ kafkaProducerProps” 使用以上配置初始化“ org.apache.storm.kafka.bolt.KafkaBolt ” 在流中包含以上KafkaBolt KafkaBolt的最小Flux配置示例 components:- id: stringSchemeclassName: org.apache.storm.kafka.StringScheme- id: stringMultiSchemeclassName: org.apache.storm.spout.SchemeAsMultiSchemeconstructorArgs:- ref: stringScheme- id: zkHostsclassName: org.apache.storm.kafka.ZkHostsconstructorArgs:- localhost:2181- id: topicSelectorclassName: org.apache.storm.kafka.bolt.selector.DefaultTopicSelectorconstructorArgs:- myTopicName- id: kafkaMapperclassName: org.apache.storm.kafka.bolt.mapper.FieldNameBasedTupleToKafkaMapper- id: kafkaProducerPropsclassName: java.util.PropertiesconfigMethods:- name: putargs:- bootstrap.servers- localhost:9092- name: putargs:- acks- 1- name: putargs:- key.serializer- org.apache.kafka.common.serialization.StringSerializer- name: putargs:- value.serializer- org.apache.kafka.common.serialization.StringSerializer bolts: - id: bolt-kafkaclassName: org.apache.storm.kafka.bolt.KafkaBoltparallelism: 1configMethods:- name: withProducerPropertiesargs: [ref: kafkaProducerProps]- name: withTopicSelectorargs: [ref: topicSelector]- name: withTupleToKafkaMapperargs: [ref: kafkaMapper]streams:- name: spout -- kafkaBoltfrom: spout-1to: bolt-kafkagrouping:type: LOCAL_OR_SHUFFLE 有关完整的工作配置示例请选中此项 可以像这样使用 。 在Storm上部署拓扑的示例命令 storm jar target/sentiment-analysis-storm-0.0.1-SNAPSHOT.jar org.apache.storm.flux.Flux --remote --c nimbus.host192.168.1.200 src/test/resources/flux/topology_kafka.yaml KafkaSpout的助焊剂配置已作为官方助焊剂示例进行了描述。 Flux是一个非常有用的框架它消除了定义和初始化拓扑所需的自定义代码 翻译自: https://www.javacodegeeks.com/2016/05/apache-storm-configure-kafkabolt-flux.html
