Apache Pulsar 内置连接器完全指南：Source 与 Sink 全清单、配置与实战

消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载Apache Pulsar 发行版内置了一组经过打包与联调验证的常见连接器Built-in Connector用于在 Pulsar 与最常用的外部数据系统数据库、消息队列、大数据组件等之间导入、导出数据。使用这些连接器只需像编写一个简单连接器一样运行它——既可以在本地运行也可以提交到 Pulsar Functions 集群。本文以官方文档 io-connectors.md 为骨架完整梳理 Pulsar 内置 Source/Sink 连接器清单并融合仓库源码与配套使用文档帮助你快速选型、配置并落地第一个数据管道。Pulsar 连接器机制速览Source 与 Sink 的基本概念Pulsar IO 连接器分为两类见 io-overview.mdSource源连接器将数据从外部系统导入 Pulsar。常见来源包括其他消息系统、Firehose 风格的数据管道 API 等。Sink汇连接器将数据从 Pulsar 导出到外部系统。常见目标包括其他消息系统以及 SQL/NoSQL 数据库。架构关系如下图所示Pulsar IO 连接器架构Source 与 Sink连接器Source 与 Sink和 Functions 一样都是实例instance的组成部分统一运行在Functions worker上。当你通过 Connector Admin CLI 或 Functions Admin CLI 管理一个 source、sink 或 function 时worker 上就会启动一个实例详见 functions-worker.md。处理保障Processing GuaranteesPulsar 连接器与 Functions 使用相同的处理保障语义用于处理向 Pulsar topic 写消息时的错误投递语义说明at-most-once每条发送给连接器的消息被处理一次或不被处理。at-least-once每条发送给连接器的消息被处理一次或多次。effectively-once每条发送给连接器的消息有且仅有一个对应的输出。连接器的处理保障不仅依赖 Pulsar 的保证还与外部系统相关即 source 与 sink 的实现SourcePulsar 保证向 Pulsar topic 写入消息时遵循处理保障这在 Pulsar 的可控范围之内。Sink处理保障取决于 sink 实现。如果 sink 实现不能以幂等方式处理重试则无法兑现处理保障。创建连接器时可设置以下处理保障语义ATLEAST_ONCE、ATMOST_ONCE、EFFECTIVELY_ONCE。若创建连接器时未指定--processing-guarantees默认语义为ATLEAST_ONCE。通过 Admin CLI 设置处理保障的示例# Source设置为 at-most-once $ bin/pulsar-admin sources create \ --processing-guarantees ATMOST_ONCE \ # Other source configs # Sink设置为 effectively-once $ bin/pulsar-admin sinks create \ --processing-guarantees EFFECTIVELY_ONCE \ # Other sink configs创建完成后还可以更新处理保障$ bin/pulsar-admin sources update \ --processing-guarantees EFFECTIVELY_ONCE \ # Other source configs $ bin/pulsar-admin sinks update \ --processing-guarantees ATMOST_ONCE \ # Other sink configsSource 连接器全清单Pulsar 内置多种 Source 连接器以下按字母序排列。每个连接器均提供官方配置文档含完整属性表与示例以及仓库内可阅读的 Java 实现类。CanalCanal Source 连接器将 MySQL 变更数据拉取到 Pulsar topic。配置文档io-canal-source.md#configuration使用示例io-canal-source.md#usageJava 实现类CanalStringSource.javaDebezium 系列Debezium Source 连接器是一组基于 Kafka Connect Debezium 生态的 CDCChange Data Capture连接器将数据库变更实时捕获并持久化到 Pulsar topic。Pulsar 内置以下 5 种数据库的 Debezium 连接器连接器配置与示例Java 实现类Debezium MySQLio-debezium-source.md#example-of-mysqlDebeziumMysqlSource.javaDebezium PostgreSQLio-debezium-source.md#example-of-postgresqlDebeziumPostgresSource.javaDebezium MongoDBio-debezium-source.md#example-of-mongodbDebeziumMongoDbSource.javaDebezium Oracleio-debezium-source.md#example-of-oracleDebeziumOracleSource.javaDebezium Microsoft SQL Serverio-debezium-source.md#example-of-microsoft-sqlDebeziumMsSqlSource.javaDebezium 系列共用的核心实现位于 pulsar-io/debezium/core其中 DebeziumSource.java 是基类PulsarDatabaseHistory.java 负责将 DDL 历史写入 Pulsar topic该 topic 仅供内部使用消费者不应订阅。DynamoDBDynamoDB Source 连接器从 AWS DynamoDB Streams 拉取数据写入 Pulsar。配置文档io-dynamodb-source.md#configurationJava 实现类DynamoDBSource.javaFileFile Source 连接器读取本地文件系统含 gzip/zip 压缩文件中的文本行写入 Pulsar topic。配置文档io-file-source.md#configuration使用示例io-file-source.md#usageJava 实现类FileSource.javaFile 连接器在 pulsar-io/file 模块中还包含 FileListingThread.java、FileConsumerThread.java、ProcessedFileThread.java 等多个线程组件分别负责目录扫描、文件读取与已处理文件归档并有配套单元测试见pulsar-io/file/src/test。FlumeFlume Source 连接器将 Apache Flume 事件导入 Pulsar。配置文档io-flume-source.md#configurationJava 实现类FlumeConnector.javaTwitter firehoseTwitter Firehose Source 连接器从 Twitter 实时数据流中拉取推文写入 Pulsar。配置文档io-twitter-source.md#configurationJava 实现类TwitterFireHose.javaKafkaKafka Source 连接器从 Kafka topic 拉取消息并持久化到 Pulsar topic是消息系统间桥接的核心组件。配置文档io-kafka-source.md#configuration使用示例io-kafka-source.md#usageJava 实现类KafkaAbstractSource.javaKinesisKinesis Source 连接器从 AWS Kinesis Streams 拉取数据写入 Pulsar。配置文档io-kinesis-source.md#configurationJava 实现类KinesisSource.javaNettyNetty Source 连接器通过 TCP 或 HTTP 协议接收字节流写入 Pulsar。配置文档io-netty-source.md#configurationTCP 示例io-netty-source.md#tcpHTTP 示例io-netty-source.md#httpJava 实现类NettySource.javaNSQNSQ Source 连接器从 NSQ 分布式消息队列拉取消息写入 Pulsar。配置文档io-nsq-source.md#configurationJava 实现类NSQSource.javaRabbitMQRabbitMQ Source 连接器从 RabbitMQ 队列/交换机拉取消息写入 Pulsar。配置文档io-rabbitmq-source.md#configurationJava 实现类RabbitMQSource.javaSink 连接器全清单Pulsar 内置多种 Sink 连接器以下按字母序排列同样为每个连接器标注配置文档与 Java 实现类。Aerospike将 Pulsar 消息写入 Aerospike 数据库。配置文档io-aerospike-sink.md#configurationJava 实现类AerospikeStringSink.javaCassandra将 Pulsar 消息写入 Apache Cassandra 表。配置文档io-cassandra-sink.md#configuration使用示例io-cassandra-sink.md#usageJava 实现类CassandraStringSink.javaElasticSearch将 Pulsar 消息写入 Elasticsearch 索引。配置文档io-elasticsearch-sink.md#configurationJava 实现类ElasticSearchSink.javaFlume将 Pulsar 消息作为 Flume 事件输出到 Flume agent。配置文档io-flume-sink.md#configurationJava 实现类StringSink.javaHBase将 Pulsar 消息写入 Apache HBase 表。配置文档io-hbase-sink.md#configurationJava 实现类HbaseAbstractConfig.javaHDFS2将 Pulsar 消息写入 HDFSHadoop 2.x文件系统。配置文档io-hdfs2-sink.md#configurationJava 实现类AbstractHdfsConnector.javaHDFS3将 Pulsar 消息写入 HDFSHadoop 3.x文件系统。配置文档io-hdfs3-sink.md#configurationJava 实现类AbstractHdfsConnector.javaInfluxDB将 Pulsar 消息写入 InfluxDB 时序数据库。配置文档io-influxdb-sink.md#configurationJava 实现类InfluxDBGenericRecordSink.javaJDBC 系列JDBC Sink 连接器将 Pulsar 消息写入关系型数据库Pulsar 内置以下 4 种数据库连接器配置与示例Java 实现类JDBC ClickHouseio-jdbc-sink.md#example-for-clickhouseClickHouseJdbcAutoSchemaSink.javaJDBC MariaDBio-jdbc-sink.md#example-for-mariadbMariadbJdbcAutoSchemaSink.javaJDBC PostgreSQLio-jdbc-sink.md#example-for-postgresqlPostgresJdbcAutoSchemaSink.javaJDBC SQLiteio-jdbc-sink.md#example-for-sqliteSqliteJdbcAutoSchemaSink.javaJDBC 系列共用的核心实现位于 pulsar-io/jdbc/core。Kafka将 Pulsar 消息写入 Kafka topic。配置文档io-kafka-sink.md#configurationJava 实现类KafkaAbstractSink.javaKinesis将 Pulsar 消息写入 AWS Kinesis Streams。配置文档io-kinesis-sink.md#configurationJava 实现类KinesisSink.javaMongoDB将 Pulsar 消息写入 MongoDB 集合。配置文档io-mongo-sink.md#configurationJava 实现类MongoSink.javaRabbitMQ将 Pulsar 消息写入 RabbitMQ。配置文档io-rabbitmq-sink.md#configurationJava 实现类RabbitMQSink.javaRedis将 Pulsar 消息写入 Redis。配置文档io-redis-sink.md#configurationJava 实现类RedisAbstractConfig.javaSolr将 Pulsar 消息写入 Apache Solr。配置文档io-solr-sink.md#configurationJava 实现类SolrSinkConfig.java使用内置连接器安装、配置与运行完整使用流程详见 io-use.md 与 io-cli.md以下是核心步骤。安装内置连接器Pulsar 发行版已捆绑上文列出的内置连接器。安装方式请参考 getting-started-standalone.md 中的说明。安装完成后内置连接器会被 Pulsar broker或 functions worker自动发现无需额外安装步骤。若要使用非内置连接器则需要指定连接器归档文件NAR 文件的路径。配置默认存储位置在 functions_worker.yml 配置文件中设置connectorsDirectory参数即可指定内置连接器的默认存储目录######################## # Connectors ######################## connectorsDirectory: ./connectors使用 YAML 文件配置连接器创建连接器时需提供一个 YAML 配置文件告诉 Pulsar 连接器位于何处、以及如何将连接器与 Pulsar topic 连接起来。以下是官方文档给出的三个典型示例。示例 1Cassandra Sink指定连接的 Cassandra 集群、用于收集数据的keyspace与columnFamily以及 Pulsar 消息如何映射为 Cassandra 表的 key 与列tenant: public namespace: default name: cassandra-test-sink ... # cassandra specific config configs: roots: localhost:9042 keyspace: pulsar_test_keyspace columnFamily: pulsar_test_table keyname: key columnName: col示例 2Kafka Sourceconfigs: bootstrapServers: pulsar-kafka:9092 groupId: test-pulsar-io topic: my-topic sessionTimeoutMs: 10000 autoCommitEnabled: false示例 3PostgreSQL JDBC Sinkconfigs: userName: postgres password: password jdbcUrl: jdbc:postgresql://localhost:5432/test_jdbc tableName: test_jdbc获取可用连接器reload如果在连接器目录中新增或删除了 NAR 文件使用前需先重载可用内置连接器$ pulsar-admin sources reload $ pulsar-admin sinks reloadavailable重载后可选可列出当前可用的连接器$ pulsar-admin sources available-sources $ pulsar-admin sinks available-sinks创建与运行连接器创建连接器可使用Admin CLI、REST API或Java Admin API三种方式操作Admin CLIREST API 端点Java Admin API创建 Sourcepulsar-admin sources createPOST /admin/v3/sources/:tenant/:namespace/:sourceNamecreateSource(SourceConfig, String fileName)/createSourceWithUrl(SourceConfig, String pkgUrl)创建 Sinkpulsar-admin sinks createPOST /admin/v3/sinks/:tenant/:namespace/:sinkNamecreateSink(SinkConfig, String fileName)/createSinkWithUrl(SinkConfig, String pkgUrl)其中 Java Admin API 的createSourceWithUrl/createSinkWithUrl支持http与file两类 URL例如http://www.repo.com/fileName.jar、file:///dir/fileName.jar来远程下载连接器包。运行方式分为start启动已创建的连接器实例Admin CLI 子命令为pulsar-admin sources start/pulsar-admin sinks startREST API 支持启动全部实例或指定:instanceId的单个实例。localrun在本地运行连接器而非部署到 Pulsar 集群pulsar-admin sources localrun/pulsar-admin sinks localrun适合开发调试。监控连接器get获取连接器配置信息pulsar-admin sources get/pulsar-admin sinks get对应 RESTGET /admin/v3/sources/:tenant/:namespace/:sourceName与GET /admin/v3/sinks/...。list列出所有运行中的连接器pulsar-admin sources list/pulsar-admin sinks list。status获取连接器当前运行状态pulsar-admin sources status/pulsar-admin sinks status支持查询全部实例或指定实例。以 Java Admin API 为例getSource返回的SourceConfig结构如下parallelism为实例并发度resources为 CPU/内存/磁盘配额默认分别为 1.0、1 GiB、10 GiB{ tenant: public, namespace: default, name: debezium-mysql-source, className: org.apache.pulsar.io.debezium.mysql.DebeziumMysqlSource, topicName: debezium-mysql-topic, configs: { ... }, parallelism: 1, processingGuarantees: ATLEAST_ONCE, resources: { cpu: 1.0, ram: 1073741824, disk: 10737418240 } }实战示例一Canal MySQL 变更捕获以官方文档 io-canal-source.md 为例展示从 MySQL 通过 Canal 向 Pulsar 同步数据的完整流程。Canal Source 配置属性名称必填默认值说明usernametrueNoneCanal server 账号注意不是 MySQL 账号passwordtrueNoneCanal server 密码不是 MySQL 密码destinationtrueNoneCanal Source 连接器连接的 destinationsingleHostnamefalseNoneCanal server 地址singlePortfalseNoneCanal server 端口clustertruefalse是否基于 Canal server 配置启用集群模式true为集群模式向zkServers查询实际数据库主机false为单机模式直连singleHostname与singlePort指定的数据库zkServerstrueNone集群模式下用于解析实际数据库主机的 ZooKeeper 地址与端口batchSizefalse1000从 Canal 拉取的批大小配置示例YAMLconfigs: zkServers: 127.0.0.1:2181 batchSize: 5120 destination: example username: password: cluster: false singleHostname: 127.0.0.1 singlePort: 11111运行步骤启动 MySQL 并开启 ROW 格式 binlog配置mysqld.cnflog-binmysql-bin、binlog-formatROW、server_id1。启动 Canal server 并连接到 MySQL通过环境变量指定canal.instance.master.address、canal.instance.dbUsername、canal.instance.dbPassword等。启动 Pulsar standalone。使用localrun方式启动 Canal Source$ ./bin/pulsar-admin source localrun \ --archive ./connectors/pulsar-io-canal-2.3.0.nar \ --classname org.apache.pulsar.io.canal.CanalStringSource \ --tenant public \ --namespace default \ --name canal \ --destination-topic-name my-topic \ --source-config-file /pulsar/conf/canal-mysql-source-config.yaml \ --parallelism 1使用 Pulsar Python 客户端消费 topic然后在 MySQL 中执行 INSERT/UPDATE/DELETE即可在 Pulsar 端实时看到变更事件。实战示例二Kafka 与 Pulsar 桥接以官方文档 io-kafka-source.md 为例将 Kafka topic 消息桥接到 Pulsar。Kafka Source 配置属性名称类型必填默认值说明bootstrapServersStringtrue空字符串建立到 Kafka 集群初始连接的 host:port 列表逗号分隔groupIdStringtrue空字符串标识消费者所属消费组的唯一字符串fetchMinByteslongfalse1每次 fetch 响应期望的最小字节数autoCommitEnabledbooleanfalsetrue是否后台周期性提交消费者 offset该 offset 在进程故障时用作新消费者的起始位置autoCommitIntervalMslongfalse5000autoCommitEnabledtrue时消费者 offset 自动提交到 Kafka 的频率毫秒heartbeatIntervalMslongfalse3000心跳间隔必须小于sessionTimeoutMssessionTimeoutMslongfalse30000检测消费者故障的超时时间topicStringtrue空字符串向 Pulsar 发送消息的 Kafka topicconsumerConfigPropertiesMapfalse空字符串传递给消费者的额外配置连接器配置文件中指定的其他属性优先级更高keyDeserializationClassStringfalseorg.apache.kafka.common.serialization.StringDeserializer反序列化 key 的类valueDeserializationClassStringfalseorg.apache.kafka.common.serialization.ByteArrayDeserializer反序列化 value 的类autoOffsetResetStringfalseearliest默认 offset 重置策略Schema 管理Kafka Source 会根据 Kafka topic 上实际存在的数据类型为 Pulsar topic 应用 Schema数据类型由keyDeserializationClass与valueDeserializationClass决定若valueDeserializationClass为StringDeserializer可在 Pulsar topic 上设置Schema.STRING()。若为 ConfluentKafkaAvroDeserializerPulsar 会从 Confluent Schema Registry 下载 AVRO Schema 并设置到 Pulsar topic此时需在consumerConfigProperties中配置schema.registry.url。若keyDeserializationClass不是StringDeserializerKafka Source 使用KeyValue Schema SEPARATED 编码Pulsar topic 的 Key 为 Kafka 消息 key 的内容base64 编码Value 为消息 value 内容KeySchema/ValueSchema 由对应的反序列化类推导。这样topic 压缩compaction与分区路由都由 Pulsar key 驱动与 Kafka 端保持一致消费时可用KeyValueSchema 解码或用Message#getKeyBytes()访问原始 key。配置示例YAMLconfigs: bootstrapServers: pulsar-kafka:9092 groupId: test-pulsar-io topic: my-topic sessionTimeoutMs: 10000 autoCommitEnabled: false运行步骤启动 ZooKeeper 与 Kafka同一 Docker 网络内。启动 Pulsar standalone同一网络。编写 Kafka 生产者脚本向my-topic发送消息。在 Pulsar 容器内执行localrun启动 Kafka Source$ ./bin/pulsar-admin source localrun \ --archive ./pulsar-io-kafka-2.4.0.nar \ --classname org.apache.pulsar.io.kafka.KafkaBytesSource \ --tenant public \ --namespace default \ --name kafka \ --destination-topic-name my-topic \ --source-config-file ./conf/kafkaSourceConfig.yaml \ --parallelism 1在 Pulsar 消费者终端即可看到Received message: hello world。源码级实现参考所有内置连接器的实现均位于 pulsar-io 模块每个连接器一个独立 Maven 子模块产出 NAR 归档文件并在 pulsar-io/pom.xml 中统一管理。除上文列出的实现类外以下位置也值得深入阅读连接器抽象与框架pulsar-io/coreSource/Sink 基类与配置基类Kafka Connect 适配层pulsar-io/kafka-connect-adaptorDebezium 等基于 Kafka Connect 的连接器依赖此适配层AWS 系列连接器的共享实现pulsar-io/aws连接器管理员命令参考reference-connector-admin.md连接器 CLI 子命令参考io-cli.md连接器开发指南面向自定义连接器io-develop.md每个连接器模块下均带有src/test单元测试例如 KafkaAbstractSourceTest.java、NettySourceConfigTest.java、PulsarDatabaseHistoryTest.java可据此验证各连接器的配置解析与核心行为。小结本文完整梳理了 Apache Pulsar 内置连接器生态Source 侧涵盖 Canal、Debezium 全家桶MySQL/PostgreSQL/MongoDB/Oracle/MSSQL、DynamoDB、File、Flume、Twitter、Kafka、Kinesis、Netty、NSQ、RabbitMQ 共 11 大类Sink 侧涵盖 Aerospike、Cassandra、ElasticSearch、Flume、HBase、HDFS2/3、InfluxDB、JDBC 四件套、Kafka、Kinesis、MongoDB、RabbitMQ、Redis、Solr 共 14 大类。结合 io-overview.md 的处理保障语义与 io-use.md 的安装配置流程你可以快速选型并搭建外部系统 ↔ Pulsar的数据管道实现数据库 CDC 捕获、跨消息系统桥接、日志落库与数据仓库同步等典型场景。赞分享消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载相关推荐Apache Pulsar 内置连接器Built-in Connectors完全指南Source/Sink 全景、配置选型与集群实战Apache Pulsar 内置连接器Built in Connectors完全指南Source/Sink 全景、配置选型与集群实战 Apache Pul消息队列后端流处理Apache Pulsar 内置连接器Built-in Connector完全指南Source 与 Sink 生态全览及实战部署Apache Pulsar 内置连接器Built in Connector完全指南Source 与 Sink 生态全览及实战部署 Pulsar 发行版中打消息队列后端流处理Apache Pulsar Flume Sink 连接器配置与使用指南Apache Pulsar Flume Sink 连接器配置与使用指南 Flume sink 连接器是 Apache Pulsar 的 IO 连接器之一负责将消息队列后端流处理创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考