内容介绍

Pulsar介绍

Pulsar关键特性

Pulsar vs Kafka

Pulsar架构设计

Pulsar消息机制

Pulsar Schema

Pulsar Functions

Pulsar Connectors

Pulsar Deployment

Pulsar Admin

Pulsar Manager

Pulsar Flink

什么是Pulsar？

Apache Pulsar 是 Apache 软件基金会顶级项目，是下一代云原生分布式消息流平台，集消息、存储、轻量化函数式计算为一体，采用计算与存储分离架构设计，支持多租户、持久化存储、多机房跨区域数据复制，具有强一致性、高吞吐以及低延时的高可扩展流数据存储特性。

Pulsar 的关键特性

• Pulsar 的单个实例原生支持多个集群，可跨机房在集群间无缝地完成消息复制。
• 极低的发布延迟和端到端延迟。
• 可无缝扩展到超过 一百万 个 topic。
• 简单的客户端 API，支持 Java、Go、Python 和 C++。
• 支持多种 topic 订阅模式(独占订阅、共享订阅、故障转移订阅)。
• 通过 Apache BookKeeper 提供的持久化消息存储机制保证消息传递 。
• 由轻量级的 serverless 计算框架 Pulsar Functions 实现流原生的数据处理。
• 基于 Pulsar Functions 的 serverless connector 框架 Pulsar IO 使得数据更易移入、移出 Apache Pulsar。
• 分层式存储可在数据陈旧时，将数据从热存储卸载到冷/长期存储(如S3、GCS)中。

Pulsar vs Kafka

下方链接为Pulsar与 Kafka详细对比报告，可自行下载查看

https://streamnative.io/en/blog/tech/2020-07-08-pulsar-vs-kafka-part-1

https://streamnative.io/zh/blog/tech/2020-07-22-pulsar-vs-kafka-part-2

• 性能与可用性

基准测试(StreamNative)

数据来源

https://mp.weixin.qq.com/s/UZJTOEpzX8foUJv9XMJxOw

https://streamnative.io/en/blog/tech/2020-11-09-benchmark-pulsar-kafka-performance

https://streamnative.io/whitepaper/benchmark-pulsar-vs-kafka

• 吞吐量(Throughput)

在与 Kafka 的持久性保证相同的情况下，Pulsar 可达到 605 MB/s 的发布和端到端吞吐量(与 Kafka 相同)以及 3.5 GB/s 的 catch-up read 吞吐量(比 Kafka 高 3.5 倍)。Pulsar 的吞吐量不会因分区数量的增加和持久性级别的改变而受到影响，而 Kafka 的吞吐量会因分区数量或持久性级别的改变而受到严重影响。

• 延迟性(Latency)

在不同的测试实例(包括不同订阅数量、不同主题数量和不同持久性保证)中，Pulsar 的延迟显著低于 Kafka。Pulsar P99 延迟在 5 到 15 毫秒之间。Kafka P99 延迟可能长达数秒，并且会因主题数量、订阅数量和不同持久性保证而受到巨大影响。

• 功能性

1. 多语言客户端(C/C++、Python、Java、Go ...)
2. 管理工具(Pulsar Manager vs Kafka Manager)
3. 内置流处理Built-In Stream Processing(Pulsar Function vs Kafka Streams)
4. Rich Integrations (Pulsar Connectors)
5. Exactly-Once Processing
6. 日志压缩
7. 多租户(Pulsar)
8. 安全管理(Pulsar)

架构设计

Pulsar采用存储和计算分离的软件架构。在消息领域，Pulsar 是第一个将存储计算分离云原生架构落地的开源项目。由于在 Broker 层不存储任何数据，这种架构为用户带来了更高的可用性、更灵活的扩容和管理、避免数据的 reblance 和 catch-up。

在 Apache Pulsar 的分层架构中，服务层 Broker 和存储层 BookKeeper 的每个节点都是对等的。Broker 仅仅负责消息的服务支持，不存储数据。这为服务层和存储层提供了瞬时的节点扩展和无缝的失效恢复。

持久化存储(Persistent storage)

Pulsar 使用 BookKeeper 分布式日志存储数据库作为存储组件，在底层使用日志作为存储模型。

Pulsar 将所有未确认消息(即未处理消息)存储在 BookKeeper 中的多个“bookie”服务器上。

BookKeeper 通过 Quorum Vote 的方式来实现数据的一致性，跟 Master/Slave 模式不同，BookKeeper 中每个节点也是对等的，对一份数据会并发地同时写入指定数目的存储节点。

一个Topic实际上是一个ledgers流。Ledger本身就是一个日志。所以一系列的子日志(Ledgers)组成了一个父日志(Topic)。

Ledgers追加到一个Topic，条目(消息或者一组消息)追加到Ledgers。Ledger一旦关闭是不可变的。Ledger作为最小的删除单元，也就是说我们不能删除单个条目而是去删除整个Ledger。

Ledgers本身也被分解为多个Fragment。Fragment是BookKeeper集群中最小的分布单元。

每个Ledger(由一个或多个Fragment组成)可以跨多个BookKeeper节点(Bookies)进行复制，以实现数据容灾和提升读取性能。每个Fragment都在一组不同的Bookies中复制(存在足够的Bookies)。

conf/bookkeeper.conf

#############################################################################
## Server parameters
#############################################################################
# Directories BookKeeper outputs its write ahead log.
# Could define multi directories to store write head logs, separated by ','.
journalDirectories=/data/appData/pulsar/bookkeeper/journal
#############################################################################
## Ledger storage settings
#############################################################################
# Directory Bookkeeper outputs ledger snapshots
# could define multi directories to store snapshots, separated by ','
ledgerDirectories=/data/appData/pulsar/bookkeeper/ledgers

conf/broker.conf

### --- Managed Ledger --- ###
# Number of bookies to use when creating a ledger
managedLedgerDefaultEnsembleSize=2
# Number of copies to store for each message
managedLedgerDefaultWriteQuorum=2
# Number of guaranteed copies (acks to wait before write is complete)
managedLedgerDefaultAckQuorum=2

元数据存储(Metadata storage)

Pulsar和BookKeeper都使用Apache Zookeeper来存储元数据和监控节点健康状况。

$ $PULSAR_HOME/bin/pulsar zookeeper-shell
> ls /
[admin, bookies, counters, ledgers, loadbalance, managed-ledgers, namespace, pulsar, schemas, stream, zookeeper]

消息机制

Pulsar 采用发布-订阅(pub-sub)的设计模式 。该设计模式中，producer 发布消息到 topic， Consumer 订阅 topic、处理发布的消息，并在处理完成后发送确认。

一旦创建订阅，即使 consumer 断开连接，Pulsar 仍然可以保存所有消息。在 consumer 确认消息已处理成功后，才会删除消息。

主题(Topic)

逻辑上一个Topic是日志结构，每个消息都在这个日志结构中有一个偏移量。Apache Pulsar使用游标来跟踪偏移量(Cursor Tracking)。

Pulsar 支持两种基本的 topic 类型：持久 topic 与非持久 topic。

{persistent|non-persistent}://tenant/namespace/topic

• Non-Partitioned topics

$ $PULSAR_HOME/bin/pulsar-admin topics \
list public/default
$ $PULSAR_HOME/bin/pulsar-admin topics \
create persistent://public/default/input-seed-avro-topic
$ $PULSAR_HOME/bin/pulsar-admin topics \
lookup persistent://public/default/input-seed-avro-topic
$ $PULSAR_HOME/bin/pulsar-admin topics \
delete persistent://public/default/input-seed-avro-topic
$ $PULSAR_HOME/bin/pulsar-admin topics \
stats persistent://public/default/input-seed-avro-topic
$ curl http://server-101:8080/admin/v2/persistent/public/default/exclamation-input/stats | python -m json.tool

Partitioned topics

$ $PULSAR_HOME/bin/pulsar-admin topics \
create-partitioned-topic persistent://public/default/output-seed-avro-topic \
--partitions 2
$ $PULSAR_HOME/bin/pulsar-admin topics \
list-partitioned-topics public/default
$ $PULSAR_HOME/bin/pulsar-admin topics \
get-partitioned-topic-metadata persistent://public/default/output-seed-avro-topic
$ $PULSAR_HOME/bin/pulsar-admin topics \
delete-partitioned-topic persistent://public/default/output-seed-avro-topic

消息(Message)

Messages are the basic "unit" of Pulsar.

public interface Message<T> {
    Map getProperties();
    boolean hasProperty(String var1);
    String getProperty(String var1);
    byte[] getData();
    T getValue();
    MessageId getMessageId();
    long getPublishTime();
    long getEventTime();
    long getSequenceId();
    String getProducerName();
    boolean hasKey();
    String getKey();
    boolean hasBase64EncodedKey();
    byte[] getKeyBytes();
    boolean hasOrderingKey();
    byte[] getOrderingKey();
    String getTopicName();
    Optional getEncryptionCtx();
    int getRedeliveryCount();
    byte[] getSchemaVersion();
    boolean isReplicated();
    String getReplicatedFrom();
}

生产者(Producer)

public void send() throws PulsarClientException {
    final String serviceUrl = "pulsar://server-100:6650";
    // final String serviceUrl = "pulsar://server-101:6650,server-102:6650,server-103:6650";
    // http://pulsar.apache.org/docs/en/client-libraries-java/#client
    final PulsarClient client = PulsarClient.builder()
            .serviceUrl(serviceUrl)
            .connectionTimeout(10000, TimeUnit.MILLISECONDS)
            .build();
    final String topic = "persistent://public/default/topic-sensor-temp";
    // http://pulsar.apache.org/docs/en/client-libraries-java/#configure-producer
    final Producer<byte[]> producer = client.newProducer()
            .producerName("sensor-temp")
            .topic(topic)
            .compressionType(CompressionType.LZ4)
            .enableChunking(true)
            .enableBatching(true)
            .batchingMaxBytes(1024)
            .batchingMaxMessages(10)
            .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS)
            .blockIfQueueFull(true)
            .maxPendingMessages(512)
            .sendTimeout(1, TimeUnit.SECONDS)
            .create();
    MessageId mid = producer.send("sensor-temp".getBytes());
    System.out.printf("\nmessage with ID %s successfully sent", mid);
    mid = producer.newMessage()
            .key("sensor-temp-key")
            .value("sensor-temp-key".getBytes())
            .property("my-key", "my-value")
            .property("my-other-key", "my-other-value")
            .send();
    System.out.printf("message-key with ID %s successfully sent", mid);
    producer.close();
    client.close();
}

消费者(Consumer)

public void consume() throws PulsarClientException {
    final String serviceUrl = "pulsar://server-101:6650";
    final String topic = "input-seed-avro-topic";
    final PulsarClient client = PulsarClient.builder()
            .serviceUrl(serviceUrl)
            .enableTcpNoDelay(true)
            .build();
    final Consumer<byte[]> consumer = client
            .newConsumer()
            .consumerName("seed-avro-consumer")
            .subscriptionName("seed-avro-subscription")
            .subscriptionType(SubscriptionType.Exclusive)
            .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
            .topic(topic)
            .receiverQueueSize(10)
            .subscribe();
    final AvroSchema schema = AvroSchema.of(SeedEvent.class);
    while (true) {
        try {
            final Message<byte[]> msg = consumer.receive();
            LOG.info("接收消息:[{}] topic:{} mid:{} sid:{} event:{} publish:{} producer:{} key:{} value:{}",
                    Thread.currentThread().getId(),
                    msg.getTopicName(),
                    msg.getMessageId(),
                    msg.getSequenceId(),
                    msg.getEventTime(),
                    msg.getPublishTime(),
                    msg.getProducerName(),
                    msg.getKey(), schema.decode(msg.getValue()));
            try {
                consumer.acknowledge(msg);
            } catch (final PulsarClientException e) {
                consumer.negativeAcknowledge(msg);
                LOG.error("acknowledge:" + e.getLocalizedMessage(), e);
            }
        } catch (final PulsarClientException e) {
            LOG.error("receive:" + e.getLocalizedMessage(), e);
        }
    }
}

订阅(Subscriptions)

消费者通过订阅来消费Topic中的消息。订阅是游标(跟踪偏移量)的逻辑实体，一个Topic可以添加多个订阅。订阅不包含消息的数据，只包含元数据和游标。

每个Subscription都存储一个Cursor。Cursor是日志中的当前偏移量。Subscription将其Cursor存储至BookKeeper的Ledger中。这使Cursor跟踪可以像Topic一样进行扩展。

订阅类型(subscription-type)

• Exclusive 独享

一个订阅只能有一个消息者消费消息。

• Failover 灾备

一个订阅同时只有一个消费者，可以有多个备份消费者。一旦主消费者故障则备份消费者接管。不会出现同时有两个活跃的消费者。

• Shared 共享

一个订阅中同时可以有多个消费者，多个消费者共享Topic中的消息。

• Key_Shared

有序性保证(Ordering guarantee)

如果对顺序性有要求，可以使用 Exclusive 和 Failover 的订阅模式，这样同一个 Topic 只有一个 Consumer 在消费，可以保证顺序性。

如果使用 Shared 订阅模式，多个 Consumer 可以并发消费同一个 Topic。通过动态增加 Consumer 的数量，可以加速 Topic 的消费，减少消息在服务端的堆积。

KeyShared 模式保证在 Shared 模式下同一个 Key 的消息也会发送到同一个 Consumer，在并发的同时也保证了顺序性。

多主题订阅(Multi-topic subscriptions)

Pattern:

• persistent://public/default/.*
• persistent://public/default/foo.*

Reader

public void read() throws IOException {
    final String serviceUrl = "pulsar://server-101:6650";
    final PulsarClient client = PulsarClient.builder()
            .serviceUrl(serviceUrl)
            .build();
    // http://pulsar.apache.org/docs/en/client-libraries-java/#reader
    final Reader<byte[]> reader = client.newReader()
            .topic("my-topic")
            .startMessageId(MessageId.earliest()) // MessageId.latest
            .create();
    while (true) 
        final Message<byte[]> message = reader.readNext();
        System.out.println(new String(message.getData()));
    }
}

分片主题(Partitioned topics)

消息保留和过期(Message retention and expiry)

如果没有对Topic设置数据保留策略，一旦一个Topic的所有订阅的游标都已经成功消费到一个偏移量时，此偏移量前面的消息就会被自动删除。

如果Topic设置了数据保留策略，已经消费确认的消息超过保留策略阈值(Topic的消息存储大小、Topic中消息保留的时间)后会被删除。

conf/broker.conf

# Default message retention time
# 默认0, 修改为3天=60*24*3
defaultRetentionTimeInMinutes=4320
# Default retention size
# 默认为0, 修改为10G
defaultRetentionSizeInMB=10240
# Default ttl for namespaces if ttl is not already configured at namespace policies. (disable default-ttl with value 0)
ttlDurationDefaultInSeconds=0

retention policy (for a namespace)

$ $PULSAR_HOME/bin/pulsar-admin namespaces \
get-retention public/default
$ curl -X GET http://server-101:8080/admin/v2/namespaces/public/default/retention | python -m json.tool
$ $PULSAR_HOME/bin/pulsar-admin namespaces \
set-retention public/default \
--size 1024M \
--time 5m
$ curl -X POST http://server-101:8080/admin/v2/namespaces/public/default/retention \
--header "Content-Type:application/json" \
--data '{
  "retentionTimeInMinutes" : 5,
  "retentionSizeInMB" : 1024
}'

message expiry / message-ttl

$ $PULSAR_HOME/bin/pulsar-admin namespaces \
get-message-ttl public/default
$ curl -X GET http://server-101:8080/admin/v2/namespaces/public/default/messageTTL
$ $PULSAR_HOME/bin/pulsar-admin namespaces \
set-message-ttl public/default \
--messageTTL 1800
$ curl -X POST http://server-101:8080/admin/v2/namespaces/public/default/messageTTL \
--header "Content-Type:application/json" \
--data '1800'

Pulsar Schema

Pulsar schema enables you to use language-specific types of data when constructing and handling messages from simple types to more complex application-specific types.

• 类型安全(序列化和反序列化)
• Schema 帮助 Pulsar 保留了数据在其他系统中原有的含义

Schema类型(Schema type)

• Primitive type

Producer producer = client.newProducer(Schema.STRING).create();
producer.newMessage().value("Hello Pulsar!").send();
Consumer consumer = client.newConsumer(Schema.STRING).subscribe();
consumer.receive();

• Complex type

1. keyvalue key/value pair.

Schema> schema = Schema.KeyValue(
Schema.INT32,
Schema.STRING,
KeyValueEncodingType.SEPARATED
);
// Producer
Producer> producer = client.newProducer(schema)
    .topic(TOPIC)
    .create();
final int key = 100;
final String value = "value-100";
producer.newMessage().value(new KeyValue<>(key, value)).send();
// Consumer
Consumer> consumer = client.newConsumer(schema)
    .topic(TOPIC).subscriptionName(SubscriptionName).subscribe();
Message> msg = consumer.receive();

2.struct AVRO, JSON, and Protobuf.

Producer producer = client.newProducer(Schema.AVRO(User.class)).create();
producer.newMessage().value(User.builder().userName("pulsar-user").userId(1L).build()).send();
Consumer consumer = client.newConsumer(Schema.AVRO(User.class)).subscribe();
User user = consumer.receive();

Schema工作方式(How does schema work)

Producer

Consumer

Schema管理(Schema manual management)

查询Schema

$ $PULSAR_HOME/bin/pulsar-admin schemas \
get persistent://public/default/spirit-avro-topic
$ $PULSAR_HOME/bin/pulsar-admin schemas \
get persistent://public/default/spirit-avro-topic \
--version=2

更新Schema

$ $PULSAR_HOME/bin/pulsar-admin schemas upload \
persistent://public/default/test-topic \
--filename $PULSAR_HOME/connectors/json-schema.json

提取Schema

$ $PULSAR_HOME/bin/pulsar-admin schemas \
extract persistent://public/default/test-topic  \
--classname com.cloudwise.modal.Packet \
--jar ~/cloudwise-pulsar-1.0.0-RELEASE.jar \
--type json
public void schemaInfo() {
    System.out.println("AvroSchema:" + AvroSchema.of(SeedEvent.class).getSchemaInfo());
    System.out.println("Schema.AVRO:" + Schema.AVRO(SeedEvent.class).getSchemaInfo());
}

删除Schema

$ $PULSAR_HOME/bin/pulsar-admin schemas \
delete persistent://public/default/spirit-avro-topic

Pulsar Functions

编程模型(Programming model)

开启Functions

1. conf/bookkeeper.conf

extraServerComponents=org.apache.bookkeeper.stream.server.StreamStorageLifecycleComponent

1. conf/broker.conf

functionsWorkerEnabled=true

1. conf/functions_worker.yml

pulsarFunctionsCluster: pulsar-cluster
numFunctionPackageReplicas: 2

窗口(window)

• windowLengthCount 每个窗口的消息数量
• slidingIntervalCount 窗口滑动后的消息数量
• windowLengthDurationMs 窗口时间
• slidingIntervalDurationMs 窗口滑动后的时间

开窗函数

public class WordCountWindowFunction implements org.apache.pulsar.functions.api.WindowFunction<String, Void> {
    @Override
    public Void process(Collection> inputs, WindowContext context) throws Exception {
        for (Record input : inputs) {
        }
        return null;
    }
}

运行函数

• 时间，滑动窗口

--user-config '{"windowLengthDurationMs":"60000", "slidingIntervalDurationMs":"1000"}'

• 时间，滚动窗口

--user-config '{"windowLengthDurationMs":"60000"}'

• 数量，滑动窗口

--user-config '{"windowLengthCount":"100", "slidingIntervalCount":"10"}'

• 数量，滚动窗口

--user-config '{"windowLengthCount":"100"}'

Java编程

pom.xml

<dependency>
    <groupId>org.apache.pulsargroupId>
    <artifactId>pulsar-clientartifactId>
    <version>${pulsar.version}version>
dependency>
<dependency>
    <groupId>org.apache.pulsargroupId>
    <artifactId>pulsar-functions-apiartifactId>
    <version>${pulsar.version}version>
dependency>
<dependency>
    <groupId>org.apache.pulsargroupId>
    <artifactId>pulsar-functions-local-runnerartifactId>
    <version>${pulsar.version}version>
dependency>

1. WordCount

public class WordCountFunction implements org.apache.pulsar.functions.api.Function<String, Void> {
    @Override
    public Void process(String input, Context context) throws Exception {
        Arrays.asList(input.split(" ")).forEach(word -> {
            String counterKey = word.toLowerCase();
            if (context.getCounter(counterKey) == 0) {
                context.putState(counterKey, ByteBuffer.wrap(ByteUtils.from(100)));
            }
            context.incrCounter(counterKey, 1);
        });
        return null;
    }
}
$ $PULSAR_HOME/bin/pulsar-admin functions create \
--broker-service-url pulsar://server-101:6650 \
--jar target/cloudwise-pulsar-functions-with-dependencies.jar \
--classname com.cloudwise.quickstart.pulsar.functions.WordCountFunction \
--tenant public \
--namespace default \
--name word-count-function \
--inputs persistent://public/default/sentences \
--output persistent://public/default/wordcount

1. 动态路由

/**
 * 基本思路是检查每条消息的内容，根据消息内容将消息路由到不同目的地。
 */
public class RoutingFunction implements org.apache.pulsar.functions.api.Function<String, String> {
    @Override
    public String process(String input, Context context) throws Exception {
        String regex = context.getUserConfigValue("regex").toString();
        String matchedTopic = context.getUserConfigValue("matched-topic").toString();
        String unmatchedTopic = context.getUserConfigValue("unmatched-topic").toString();
        Pattern pattern = Pattern.compile(regex);
        Matcher matcher = pattern.matcher(input);
        if (matcher.matches()) {
            context.newOutputMessage(matchedTopic, Schema.STRING).value(input).send();
        } else {
            context.newOutputMessage(unmatchedTopic, Schema.STRING).value(input).send();
        }
        return null;
    }
}

1. log-topic

public class LoggingFunction implements org.apache.pulsar.functions.api.Function<String, Void> {
    @Override
    public Void process(String s, Context context) throws Exception {
        Logger LOG = context.getLogger();
        String messageId = context.getFunctionId();
        if (s.contains("danger")) {
            LOG.warn("A warning was received in message {}", messageId);
        } else {
            LOG.info("Message {} received\nContent: {}", messageId, s);
        }
        return null;
    }
}
$ $PULSAR_HOME/bin/pulsar-admin functions create \
--jar cloudwise-pulsar-functions-1.0.0.jar \
--classname com.cloudwise.quickstart.pulsar.functions.LoggingFunction \
--log-topic persistent://public/default/logging-function-logs

1. user-config

public class UserConfigFunction implements org.apache.pulsar.functions.api.Function<String, Void> {
    @Override
    public Void process(String s, Context context) throws Exception {
        Logger log = context.getLogger();
        Optional