如何使用 Clojure API 开发 Storm 拓扑？

推荐答案

// www.javascriptcn.com code example
(ns storm-topology
  (:import [backtype.storm StormSubmitter LocalCluster]
           [backtype.storm.topology TopologyBuilder]
           [backtype.storm.tuple Fields]
           [backtype.storm.spout SchemeAsMultiScheme]
           [backtype.storm.kafka KafkaSpout SpoutConfig]
           [backtype.storm.kafka.bolt KafkaBolt]
           [backtype.storm.kafka.bolt.mapper FieldNameBasedTupleToKafkaMapper]
           [backtype.storm.kafka.bolt.selector DefaultTopicSelector]
           [backtype.storm.utils Utils]))

(defn build-topology []
  (let [builder (TopologyBuilder.)
        zk-conn-string "localhost:2181"
        kafka-topic "test-topic"
        spout-config (SpoutConfig. (ZkHosts. zk-conn-string)
                                   kafka-topic
                                   "/storm"
                                   "storm-consumer")
        kafka-spout (KafkaSpout. spout-config)
        kafka-bolt (KafkaBolt.)]
    (.setSpout builder "kafka-spout" kafka-spout)
    (.setBolt builder "kafka-bolt" kafka-bolt
              (Fields. ["key" "message"]))
    (.withTopicSelector kafka-bolt (DefaultTopicSelector. kafka-topic))
    (.withTupleToKafkaMapper kafka-bolt (FieldNameBasedTupleToKafkaMapper.))
    (.createTopology builder)))

(defn -main [& args]
  (if args
    (StormSubmitter/submitTopology "kafka-topology" {} (build-topology))
    (let [cluster (LocalCluster.)]
      (.submitTopology cluster "kafka-topology" {} (build-topology))
      (Utils/sleep 10000)
      (.shutdown cluster))))

本题详细解读

1. 导入必要的类

在Clojure中，我们首先需要导入Storm和Kafka相关的类。这些类包括TopologyBuilder、KafkaSpout、KafkaBolt等，用于构建和配置Storm拓扑。

2. 定义拓扑构建函数

build-topology函数用于构建Storm拓扑。在这个函数中，我们创建了一个TopologyBuilder实例，并配置了Kafka Spout和Kafka Bolt。

• Kafka Spout: 用于从Kafka主题中读取数据。我们通过SpoutConfig配置Zookeeper连接字符串、Kafka主题、以及消费者组信息。
• Kafka Bolt: 用于将处理后的数据写回到Kafka。我们通过DefaultTopicSelector和FieldNameBasedTupleToKafkaMapper来配置Bolt的输出主题和消息映射。

3. 设置Spout和Bolt

使用TopologyBuilder的setSpout和setBolt方法将Spout和Bolt添加到拓扑中。Spout负责从Kafka读取数据，Bolt负责将处理后的数据写回Kafka。

4. 提交拓扑

在-main函数中，我们根据是否有命令行参数来决定是本地运行拓扑还是提交到远程集群。如果有参数，则使用StormSubmitter/submitTopology提交拓扑到远程集群；否则，使用LocalCluster在本地运行拓扑。

5. 运行拓扑

在本地运行拓扑时，我们让拓扑运行10秒钟，然后关闭本地集群。

通过以上步骤，我们可以在Clojure中使用Storm API开发一个简单的Kafka拓扑。