流式写入管道

本文讨论 Ozone 中的流式写入管线（Streaming Write Pipeline）功能。该功能基于 Ratis Streaming API 实现。

写入管线版本

请注意，现有的 Ozone 写入管线是基于 Ratis Async API 实现的。我们将新的流式写入管线称为写入管线 V2，将现有的异步写入管线称为写入管线 V1。

流式写入管线 V2 通过提供更好的网络拓扑感知并消除 V1 中的性能瓶颈来提升性能。V2 实现还避免了不必要的缓冲区拷贝（通过 Netty 零拷贝），并对每个数据节点的 CPU 和磁盘进行了更充分的利用。

有关写入管线的详细架构信息，请参阅写入管线文档。

配置属性

请将以下属性设置到 Ozone 配置文件 ozone-site.xml 中。

启用流式写入管线

要启用流式写入管线功能，请将以下属性设置为 true：

<property>
  <name>hdds.container.ratis.datastream.enabled</name>
  <value>true</value>
  <description>Enable data stream of container</description>
</property>

配置 Datastream 端口

Datanode 监听以下端口以接收流式传输流量：

<property>
  <name>hdds.container.ratis.datastream.port</name>
  <value>9855</value>
  <description>The datastream port number of container</description>
</property>

启用文件系统流式写入

要在 FileSystem API 中使用 Streaming，请将以下属性设置为 true：

<property>
  <name>ozone.fs.datastream.enabled</name>
  <value>true</value>
  <description>
    Enable filesystem write via ratis streaming.
  </description>
</property>

客户端 API

OzoneDataStreamOutput

新的 OzoneDataStreamOutput 类与现有的 OzoneOutputStream 类非常相似，唯一的区别在于 OzoneDataStreamOutput 在 write 方法中使用 ByteBuffer 作为参数，而 OzoneOutputStream 使用 byte[]。采用 ByteBuffer 而非 byte[] 的原因是为了支持零缓冲区拷贝。下面展示了一个典型的 write 方法：

OzoneDataStreamOutput：

public void write(ByteBuffer b, int off, int len) throws IOException;

OzoneOutputStream：

public void write(byte[] b, int off, int len) throws IOException;

使用 ByteBuffer 可以实现零拷贝操作，从而降低 CPU 开销并提升吞吐量。

OzoneBucket

为了通过流式写入管道创建键（key），OzoneBucket 新增了以下方法。

createStreamKey

public OzoneDataStreamOutput createStreamKey(String key, long size)
    throws IOException;

public OzoneDataStreamOutput createStreamKey(String key, long size,
    ReplicationConfig replicationConfig, Map<String, String> keyMetadata)
    throws IOException;

public OzoneDataStreamOutput createStreamKey(String key, long size,
    ReplicationConfig replicationConfig, Map<String, String> keyMetadata,
    Map<String, String> tags) throws IOException;

createMultipartStreamKey

用于分段上传：

public OzoneDataStreamOutput createMultipartStreamKey(String key, long size,
    int partNumber, String uploadID) throws IOException;

请注意，上述方法的参数列表与现有的 createKey 和 createMultipartKey 方法相同。

示例

下面是一个使用内存映射缓冲区从本地文件创建 key 的示例：

// Create a memory-mapped buffer from a local file:
final FileChannel channel = ...  // local file channel
final long length = ...          // length of the data
final ByteBuffer mapped = channel.map(FileChannel.MapMode.READ_ONLY, 0, length);

// Create an OzoneDataStreamOutput
final OzoneBucket bucket = ...   // an Ozone bucket
final String key = ...           // the key name
final OzoneDataStreamOutput out = bucket.createStreamKey(key, length);

// Write the memory-mapped buffer to the key output
out.write(mapped);

// close
out.close();      // In practice, use try-with-resource to close it.
channel.close();  // In practice, use try-with-resource to close it.

参考资料

写入管线 - V1 和 V2 管线的架构概述
Java 客户端 API - Ozone Java 客户端的通用文档
基于 Ratis Streaming 的 Ozone 写入管线 V2 - 技术博客文章