如何开发一个 Pulsar Connectors ？

客官°小女子只卖身不卖艺 2024-04-01 16:10 52阅读 0赞

**目录**

1、开发 Source 连接器

（1）实现 Source 接口

（2）处理 schema 信息

2、开发 Sink 连接器

（1）实现 Sink 接口

（2）处理 schema 信息

3、Test 测试建议

4、Package 打包的两种方式

（1）NAR 打包方式

（2）uber JAR 打包方式

5、Monitor 如何对连接器进行监控？

--------------------

Pulsar 连接器可以实现 Pulsar 与其他系统之间进行数据迁移。

Pulsar 连接器是一个特殊的 [Pulsar Functions][]，所以创建一个连接器就像创建一个 Pulsar Function 一样。

Pulsar 连接器有两种类型：

<table style="width:500px;"> 
 <tbody> 
  <tr> 
   <td>类型</td> 
   <td>描述</td> 
   <td>例子</td> 
  </tr> 
  <tr> 
   <td>Source</td> 
   <td>从其他系统向 Pulsar 输入数据</td> 
   <td>RabbitMQ Source 连接器将一个 RabbitMQ 队列的消息导入到一个 Pulsar 主题。</td> 
  </tr> 
  <tr> 
   <td>Sink</td> 
   <td>从 Pulsar 向其他系统输入数据</td> 
   <td>Kinesis Sink 连接器将 Pulsar 主题的消息导出到 Kinesis 流。</td> 
  </tr> 
 </tbody> 
</table>

### 1、开发 Source 连接器 ###

#### （1）实现 Source 接口 ####

开发 Source 连接器需要实现 [Source 连接器的接口][Source]，这意味着你需要实现 open() 方法和 read() 方法。Source 接口的内容如下所示：

@InterfaceAudience.Public
    @InterfaceStability.Stable
    public interface Source<T> extends AutoCloseable {
        /**
         * 打开带有配置的连接器。
         */
        void open(Map<String, Object> config, SourceContext sourceContext) throws Exception;
    
        /**
         * 从source读取下一条消息。
         * 如果source中没有消息, 该调用会被阻塞
         */
        Record<T> read() throws Exception;
    }

**实现 open() 方法：**

/**
         * Open connector with configuration.
         *
         * @param config initialization config
         * @param sourceContext environment where the source connector is running
         * @throws Exception IO type exceptions when opening a connector
         */
        void open(Map<String, Object> config, SourceContext sourceContext) throws Exception;

当 Source 连接器被初始化时，会调用 open() 方法。// 两个参数 config 和 SourceContext

在这个方法中，你可以通过 config 参数获取连接器所有的具体配置，同时初始化所有必须的资源。

比如，一个 Kafka 连接器可以在 open() 方法中创建一个 Kafka 客户端。

此外，Pulsar 提供了一个 SourceContext 容器（文本域），该容器可以让 Pulsar 在运行期间能够访问容器的运行时资源，比如，有些任务需要收集一些运行时的参量数据。实现和保存 SourceContext ，可以方便后续有需要的时候的使用。

**实现 read() 方法：**

/**
         * Reads the next message from source.
         * If source does not have any new messages, this call should block.
         * @return next message from source.  The return result should never be null
         * @throws Exception
         */
        Record<T> read() throws Exception;

如果该方法没有返回值，那么该方法应该是被阻塞而不是直接返回 Null。

返参 [Record][] 封装很多信息，这些信息是 Pulsar IO 运行时所必须的。

**Record 需要提供以下参数：**

<table style="width:500px;"> 
 <tbody> 
  <tr> 
   <td></td> 
   <td>参数名称</td> 
   <td>是否必须</td> 
   <td>描述</td> 
  </tr> 
  <tr> 
   <td>1</td> 
   <td>TopicName</td> 
   <td>NO</td> 
   <td>数据来源的 Pulsar 主题名称</td> 
  </tr> 
  <tr> 
   <td>2</td> 
   <td>Key</td> 
   <td>NO</td> 
   <td> <p>消息可以选择性地用 Key 标记。</p> <p>有关使用 Key 的更多信息，请参考<a href="https://pulsar.apache.org/docs/concepts-messaging#routing-modes" title="这里" rel="nofollow">这里</a>。</p> </td> 
  </tr> 
  <tr> 
   <td>3</td> 
   <td>Value</td> 
   <td>Yes</td> 
   <td>Record 的实际数据</td> 
  </tr> 
  <tr> 
   <td>4</td> 
   <td>EventTime</td> 
   <td>NO</td> 
   <td>数据从 Source 读取的时间</td> 
  </tr> 
  <tr> 
   <td>5</td> 
   <td>PartitionId</td> 
   <td>NO</td> 
   <td> <p>分区 Id，如果读取的数据源是一个分区的主题，那么会返回该主题的分区 Id。</p> <p><span style="color:#e69c99;">Pulsar IO 运行时会使用 PartitionId 作为数据唯一标识符的一部分</span>，用来实现消息去重，和保证消息能被精确的一次处理。</p> </td> 
  </tr> 
  <tr> 
   <td>6</td> 
   <td>RecordSequence&nbsp;&nbsp;</td> 
   <td>NO</td> 
   <td> <p>如果数据的来源是一个序列化的&nbsp;Source，将会返回该数据的 RecordSequence&nbsp;</p> <p><span style="color:#e69c99;">Pulsar IO 运行时会使用 RecordSequence&nbsp;作为数据唯一标识符的一部分</span>，用来实现消息去重，和保证消息能被精确的一次处理。</p> </td> 
  </tr> 
  <tr> 
   <td>7</td> 
   <td>Properties</td> 
   <td>NO</td> 
   <td>如果数据携带了一些用户定义的属性，该变量将返回这些属性。</td> 
  </tr> 
  <tr> 
   <td>8</td> 
   <td>DestinationTopic&nbsp;</td> 
   <td>NO</td> 
   <td>目标主题，需要将消息写入的主题。</td> 
  </tr> 
  <tr> 
   <td>9</td> 
   <td>Message</td> 
   <td>NO</td> 
   <td> <p>用户发送的数据/消息</p> <p>关于 Message 类的更多信息，请参考<a href="https://github.com/apache/pulsar/blob/master/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Message.java" title="这里" rel="nofollow">这里</a>。</p> </td> 
  </tr> 
 </tbody> 
</table>

**Record 需要提供以下两个方法：**

<table style="width:500px;"> 
 <tbody> 
  <tr> 
   <td>方法</td> 
   <td>描述</td> 
  </tr> 
  <tr> 
   <td>ack</td> 
   <td>确认记录（消息）已被完全处理</td> 
  </tr> 
  <tr> 
   <td>fail</td> 
   <td>表示该记录（消息）处理失败</td> 
  </tr> 
 </tbody> 
</table>

#### （2）处理 schema 信息 ####

Pulsar IO 可以自动处理 schema 信息，并且提供了基于Java 泛型的强类型 API 。如果事先知道要生成的 schema 类型，就可以在 Sink 连接器（数据接收方）的声明中使用这个 schema（该 schema 使用 Java class 进行描述）。// 下边的 schema 是 String 类型

public class MySource implements Source<String> {
        public Record<String> read() {}
    }

如果你想实现的 Source 连接器能处理任何的 schema 数据类型，你可以使用 byte\[\] (of ByteBuffer) 和 Schema.AUTO\_PRODUCE\_BYTES() ：// 使用字节流

public class MySource implements Source<byte[]> {
        public Record<byte[]> read() {
            
            Schema wantedSchema = ....
            Record<byte[]> myRecord = new MyRecordImplementation(); 
            ....
        }
        class MyRecordImplementation implements Record<byte[]> {
             public byte[] getValue() {
                return ....encoded byte[]...that represents the value 
             }
             public Schema<byte[]> getSchema() {
                 return Schema.AUTO_PRODUCE_BYTES(wantedSchema);
             }
        }
    }

如果需要正确的处理 KeyValue 类型，实现 Record 时请遵循以下指南：

*  必须实现 Record 接口，同时提供 getKeySchema()、getValueSchema() 和 getKeyValueEncodingType() 等方法的实现逻辑。// 以上方法是自定义的，不是 Record 接口自带的方法
 *  必须返回一个 KeyValue 对象作为调用 Record.getValue() 的返回值。
 *  调用 Record.getSchema() 方法时可能返回 Null。// 对这种情况需要额外考虑

当 Pulsar IO 在运行期间遇到 KVRecord 类型的数据时，连接器会自动的进行以下处理：

*  为数据设置正确的 KeyValueSchema
 *  根据 KeyValueEncoding 对消息的 Key 和消息的 Value 进行编码处理

有关如何创建源连接器的更多信息，请参考 [KafkaSource][]。

// 对于如何创建一个 Source 连接器，上边给出了简单的概述和大概的步骤，比较粗犷，但是也比较核心。

### 2、开发 Sink 连接器 ###

#### （1）实现 Sink 接口 ####

开发 Sink 连接器和开发 Source 连接器类似，开发 Sink 连接器需要实现 Sink 接口。这意味着你需要实现 open() 方法和 write() 方法。Sink 接口的内容如下所示：

/**
     * Generic sink interface users can implement to run Sink on top of Pulsar Functions.
     */
    @InterfaceAudience.Public
    @InterfaceStability.Stable
    public interface Sink<T> extends AutoCloseable {
        /**
         * 打开带有配置的连接器
         */
        void open(Map<String, Object> config, SinkContext sinkContext) throws Exception;
    
        /**
         * 把消息写入 Sink
         *
         * @param record record to write to sink
         * @throws Exception
         */
        void write(Record<T> record) throws Exception;
    }

在实现 write() 方法的时候，你可以决定如何将 Value 和 Key 写入实际的 Source，并可以利用所有提供的信息，比如 PartitionId 和 RecordSequence，实现不同的处理保证（即是否消费一次还是多次，怎样保证消息发送和处理之间的幂等性等等）。

你还需要实现消息确认（如果消息发送成功）或者失败否认（如果消息发送失败）等机制。

#### （2）处理 schema 信息 ####

public class MySink implements Sink<String> {
        public void write(Record<String> record) {}
    }

如果你想实现的 Sink 连接器能处理任何的 schema 数据类型，你可以实现特殊的通用对象（GenericObject）接口：

如果数据格式是 AVRO，JSON 和 Protobuf（ schemaType = AVRO，JSON，PROTOBUF\_NATIVE）等类型的情况下，你可以把 genericObject 对象强制转换成 GenericRecord 类型，并且可以使用 getFields() 和 getField() API 。你可以通过如下调用 genericObject.getNativeObject() ，访问本地的 AVRO 格式的记录（消息/数据）。// 对于以上的 schema 类型，需要参考 Pulsar 的 schema 相关说明

如果消息是 KeyValue 类型的情况下，你可以通过如下代码，访问消息 Key 的 schema 和 Value 的 schema：

public class MySink implements Sink<GenericObject> {
        public void write(Record<GenericObject> record) {
            Schema schema = record.getSchema();
            GenericObject genericObject = record.getValue();
            SchemaType type = genericObject.getSchemaType();
            Object nativeObject = genericObject.getNativeObject();
            if (type == SchemaType.KEY_VALUE) {
                KeyValue keyValue = (KeyValue) nativeObject;
                Object key = keyValue.getKey();
                Object value = keyValue.getValue();
            
                KeyValueSchema keyValueSchema = (KeyValueSchema) schema;
                Schema keySchema = keyValueSchema.getKeySchema();
                Schema valueSchema = keyValueSchema.getValueSchema();
            }
            ....
        }
    }

### 3、Test 测试建议 ###

Pulsar IO 连接器的测试是一件比较麻烦的事情，因为连接器需要与两个系统进行交互（Pulsar 和 另一个系统），这种场景难以模拟。

所以，在模拟外部服务的同时，推荐按照如下步骤，编写特殊的测试程序来测试连接器的功能：

1.  单元测试（Unit test）：首先对连接器进行单元测试
2.  集成测试（Integration test）：当你为连接器编写了足够的单元测试，就可以添加单独的集成测试来验证端到端功能。// 先单元测试，再集成测试

可以使用[在线环境][Link 1]，完成集成测试。

有关更多 Pulsar 连接器集成测试的信息，请[点击这里][Link 2]。

### 4、Package 打包的两种方式 ###

当你开发和测试完 Pulsar 连接器后，你需要把代码进行打包，然后把他提交到一个 Pulsar Functions 的节点上。

Pulsar Functions 在运行时可以接受两种打包的方式，即 NAR 和 uber JAR。

#### （1）NAR 打包方式 ####

NAR 是 NiFi Archive 的缩写，它是 Apache NiFi 使用的一种自定义打包机制，用于提供少量的 Java ClassLoader 隔离。// 什么是少量的隔离？

Pulsar 使用相同的 NAR 的机制打包所有的内置连接器。

打包 Pulsar 连接器最简单的方法是使用 nifi-nar-maven-plugin 插件来创建 NAR 包。

如下所示，在连接器的 maven 项目中包添加 nifi-nar-maven-plugin 依赖。

<plugins>
      <plugin>
        <groupId>org.apache.nifi</groupId>
        <artifactId>nifi-nar-maven-plugin</artifactId>
        <version>1.2.0</version>
      </plugin>
    </plugins>

同时你也必须要根据以下内容，创建一个 resources/META-INF/services/pulsar-io.yaml 文件

name: connector name
    description: connector description
    sourceClass: fully qualified class name (only if source connector)
    sinkClass: fully qualified class name (only if sink connector)

#### （2）uber JAR 打包方式 ####

另一种替代方案是使用 uber JAR 打包方式，该方式包含了连接器的所有 JAR 文件和其他资源文件。该方式不需要使用文件目录的内部结构。

你可以使用 maven-shade-plugin 插件来创建 uber JAR 包：

<plugin>
      <groupId>org.apache.maven.plugins</groupId>
      <artifactId>maven-shade-plugin</artifactId>
      <version>3.1.1</version>
      <executions>
        <execution>
          <phase>package</phase>
          <goals>
            <goal>shade</goal>
          </goals>
          <configuration>
            <filters>
              <filter>
                <artifact>*:*</artifact>
              </filter>
            </filters>
          </configuration>
        </execution>
      </executions>
    </plugin>

### 5、Monitor 如何对连接器进行监控？ ###

Pulsar 连接器使你能够轻松地将数据移进和移出 Pulsar 。所以，在任何时候都需要确认 Pulsar 连接器在正常运行，你可以通过以下方式监测正在运行中的 Pulsar 连接器：

（1）检查 Pulsar 提供的相关指标。Pulsar 连接器公开了很多指标，这些指标能够被收集和用来监测连接器的健康情况。详细的监测指标，请参考[官方文档][Link 3]。

（2）设置并检查自定义的 Pulsar 指标。除了 Pulsar 提供的指标以外，Pulsar 还允许用户为 Java 连接器自定义指标。 Function workers 会自动向 Prometheus 收集用户自定义的指标，然后，你可以在 Grafana 中检验它们。

下面是如何为 Java 连接器自定义指标的示例：

public class TestMetricSink implements Sink<String> {
    
            @Override
            public void open(Map<String, Object> config, SinkContext sinkContext) throws Exception {
                sinkContext.recordMetric("foo", 1); // 自定义指标
            }
    
            @Override
            public void write(Record<String> record) throws Exception {
    
            }
    
            @Override
            public void close() throws Exception {
    
            }
        }

至此，自定义指标介绍完毕。

[Pulsar Functions]: https://pulsar.apache.org/docs/functions-overview
[Source]: https://github.com/apache/pulsar/blob/master/pulsar-io/core/src/main/java/org/apache/pulsar/io/core/Source.java
[Record]: https://github.com/apache/pulsar/blob/master/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Record.java
[KafkaSource]: https://github.com/apache/pulsar/blob/master/pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaAbstractSource.java
[Link 1]: https://www.testcontainers.org/
[Link 2]: https://github.com/apache/pulsar/tree/master//tests/integration/src/test/java/org/apache/pulsar/tests/integration/io
[Link 3]: https://pulsar.apache.org/docs/deploy-monitoring