通过 Amazon S3 为 Apache Pulsar 配置分层存储

原作者：Ivan Kelly
翻译：StreamNative-Sijia

Apache Pulsar 的分层存储特性使 Pulsar 可以将 topic 上较旧的消息转移到长期存储系统中，从而释放 Apache BookKeeper 的空间，并利用可扩展的低成本存储（如：云存储）。

分层存储对于想要长时间保留数据的 topic 益处良多。例如，一个用于训练推荐系统的存有用户操作的 topic，可能需要长时间存储数据，以便在创建新的推荐模型时，可以运行完整的用户数据。

Apache Pulsar 如何存储数据

Apache Pulsar 使用分片的架构存储 topic。Pulsar 中的 topic 会持久化到存储在 Apache BookKeeper 中的日志（即托管 ledger ）中。托管 ledger 由有序的分片列表组成。由于日志的仅追加属性，Pulsar 的写入只能作为日志的最后一个分片。所有以前的分片都被封装，且分片内的数据不可变。

分层存储卸载机制利用这种分片的架构。当分片被卸载到外部存储系统时，日志中的分片被一对一复制到该存储系统。除了正在写入的分片外，其他分片都可以被卸载。

分层存储示意图

Apache Pulsar 目前支持多个云存储系统用于分层存储。本文将通过一个简单的示例来介绍如何配置独立的 Pulsar 集群以使用 Amazon S3 来存储已卸载的分片。配置过程主要包括以下几个步骤：

在 Amazon S3 中创建并配置 bucket。
将 Pulsar 配置为使用 S3 bucket 作为存储层。
启动 Pulsar 并验证卸载。

本文末尾附有讲解视频。

配置步骤

第 1 步：设置 S3 bucket

创建将会用作分层存储的 S3 bucket。登录到 AWS 控制台，然后选择 S3 服务。

使用 AWS 控制台创建 Amazon S3 bucket

创建一个 bucket。首先，单击“创建 bucket”；然后给新创建的 bucket 命名；再点击“下一步”，确认创建操作。

创建一个 Amazon S3 bucket

这样，就创建好了一个新的 Bucket。

Amazon S3 bucket 创建成功

、

确保 AWS 凭证设置正确。

$ cat ~/.aws/credentials
[default]
aws_access_key_id = XXXXXXXXXXXXXXXXXXXXXX
aws_secret_access_key = XXXXXXXXXXXXXXXXXXXXXXXXXXX

第 2 步：配置 Pulsar

现在要将 Pulsar 配置为使用 S3 bucket 作为其冷存储层。

从 http://pulsar.apache.org/en/download 下载 Pulsar 文件并解压缩。
使用 cd 命令进入二进制根目录，编辑 conf/standalone.conf，在 conf/standalone.conf 文件末尾添加卸载配置设置：
```
managedLedgerOffloadDriver=S3
s3ManagedLedgerOffloadBucket=offload-test-aws
s3ManagedLedgerOffloadRegion=us-east-1
```

在此配置文件中，更改 ledger 大小和切换时间（rollover time），简化主题创建分片的操作：

# Max number of entries to append to a ledger before triggering a rollover
# A ledger rollover is triggered on these conditions
# * Either the max rollover time has been reached
# * or max entries have been written to the ledger and at least min-time has passed
managedLedgerMaxEnteriesPerLedger=1000

# Minimum time between ledger rollover for a topic
managedLedgerMinLedgerRolloverTimeMinutes=0

以独立模式启动 Pulsar：
```
$ bin/pulsar standalone
```

第 3 步：验证分片卸载

通过 consume 消息和 produce 消息来验证配置。

在新的终端窗口，运行 consume 命令以确保 topic 数据不会被自动删除。
```
$ bin/pulsar-client consume -s “my-sub-name“ my-topic-for-offload
```
在另一个新的终端窗口，运行 produce 命令两次以确保其大小足够创建两个分片（每个分片上有 1000 条 entry）。
```
$ bin/pulsar-client produce my-topic-for-offload  --messages "hello pulsar this is the content for each message" -n 1000
```

通过 Pulsar admin 命令行界面手动触发卸载。

$ bin/pulsar-admin topics offload --size-threshold 10K public/default/my-topic-for-offload

Offload triggered for persistent://public/default/my-topic-for-offload for messages before 32:0:-1

通过命令行界面获取卸载状态。

$ bin/pulsar-admin topics offload-status public/default/my-topic-for-offload

Offload was a success

状态为“卸载成功“后，可以使用 AWS 控制台在 S3 中找到已卸载的 ledger。

存储在 S3 中的已卸载分片文件

如果使用 Pulsar admin 命令获取 topic 内部统计信息，就会发现 ledger-31 处于“offloaded: true”状态。

$    bin/pulsar-admin topics stats-internal  public/default/my-topic-for-offload

[
  "entriesAddedCounter" : 3200,
  "numberOfEntries" : 1200,
  "totalSize" : 111344,
  "currentLedgerEntries" : 200,
  "currentLedgerSize" : 18600,
  "lastLedgerCreatedTimestamp" : "2018-10-11T07:06:14.891+08:00",
  "waitingCursorsCount" : 0,
  "pendingAddEntriesCount" : 0,
  "lastConfirmedEntry" : "32:199",
  "state" : "LedgerOpened",
  "ledgers" : [ {
      "ledgerId" : 31,
      "entries" : 1000,
      "size" : 92744,
      "offloaded" : true
  }, {
    "ledgerId" : 32,
    "entries" : 0,
    "size" : 0,
    "offloaded" : false
  } ],
  "cursors" : false

视频示范

本视频逐步介绍了如何配置完整的 Pulsar 集群以使用 Amazon S3 中的分层存储：https://www.youtube.com/embed/F0nxrjXLSxk