Pulsar adaptor for Apache Storm

Pulsar Storm is an adaptor for integrating with Apache Storm topologies. It provides core Storm implementations for sending and receiving data.

应用程序可以通过通用的 Pulsar spout 将数据写入到 Storm 拓扑中，也可以利用通用的 Pulsar bolt 来使用来自 Storm 拓扑的数据。

使用 Pulsar Storm Adaptor

引入 Pulsar Storm Adaptor 依赖:

<dependency>
  <groupId>org.apache.pulsar</groupId>
  <artifactId>pulsar-storm</artifactId>
  <version>${pulsar.version}</version>
</dependency>

Pulsar Spout

Pulsar Spout 允许 Storm topology 消费发布到 topic 的数据。 它根据收到的消息和客户端提供的 MessageToValuesMapper 发出 Storm tuple。

未能通过 downstream bolts 处理的元组将由 spout 以可配置的超时（默认为60秒）或可配置的重试次数（以先到者为准）中的指数回退量进行指数回退，然后 consumer 予以承认。 下面是 spout 的示例构造：

MessageToValuesMapper messageToValuesMapper = new MessageToValuesMapper() {
    @Override
    public Values toValues(Message msg) {
        return new Values(new String(msg.getData()));
    }
    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        // declare the output fields
        declarer.declare(new Fields("string"));
    }
};
// Configure a Pulsar Spout
PulsarSpoutConfiguration spoutConf = new PulsarSpoutConfiguration();
spoutConf.setServiceUrl("pulsar://broker.messaging.usw.example.com:6650");
spoutConf.setTopic("persistent://my-property/usw/my-ns/my-topic1");
spoutConf.setSubscriptionName("my-subscriber-name1");
spoutConf.setMessageToValuesMapper(messageToValuesMapper);
// Create a Pulsar Spout
PulsarSpout spout = new PulsarSpout(spoutConf);

点击这里查看完整示例.

Pulsar Bolt

The Pulsar bolt allows data in a Storm topology to be published on a topic. It publishes messages based on the Storm tuple received and the TupleToMessageMapper provided by the client.

A partitioned topic can also be used to publish messages on different topics. In the implementation of the TupleToMessageMapper, a “key” will need to be provided in the message which will send the messages with the same key to the same topic. Here’s an example bolt:

TupleToMessageMapper tupleToMessageMapper = new TupleToMessageMapper() {
    @Override
    public TypedMessageBuilder<byte[]> toMessage(TypedMessageBuilder<byte[]> msgBuilder, Tuple tuple) {
        String receivedMessage = tuple.getString(0);
        // message processing
        String processedMsg = receivedMessage + "-processed";
        return msgBuilder.value(processedMsg.getBytes());
    }
    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        // declare the output fields
    }
};
// Configure a Pulsar Bolt
PulsarBoltConfiguration boltConf = new PulsarBoltConfiguration();
boltConf.setServiceUrl("pulsar://broker.messaging.usw.example.com:6650");
boltConf.setTopic("persistent://my-property/usw/my-ns/my-topic2");
boltConf.setTupleToMessageMapper(tupleToMessageMapper);
// Create a Pulsar Bolt
PulsarBolt bolt = new PulsarBolt(boltConf);