跳到主要内容
版本：Next

Pulsar

Pulsar Sink 连接器

引擎支持​

Spark
Flink
SeaTunnel Zeta

主要特性​

描述​

Pulsar Sink 用于将 SeaTunnel 数据写入 Apache Pulsar topic。它既可以写入一个固定 topic，也可以在多表写入时根据每行数据携带的表标识路由到不同 topic。

支持的数据源信息​

数据源支持的版本
PulsarUniversal

输出选项​

名称类型是否必须默认值描述
topicString否-目标 Pulsar topic。普通单表写入必须配置；多表数据带有表标识时可以省略。
client.service-urlString是-Pulsar 客户端服务地址，例如 pulsar://localhost:6650。
admin.service-urlString是-Pulsar 管理端 HTTP 地址，例如 http://localhost:8080。
auth.plugin-classString否-Pulsar 认证插件类名。
auth.paramsString否-认证插件参数。需要和 auth.plugin-class 一起配置。
formatString否json数据格式。默认格式为 json。可选 text 和 avro 格式。
field_delimiterString否,当 format = "text" 时使用的字段分隔符。
semanticsEnum否AT_LEAST_ONCE写入一致性语义。可选值：NON、AT_LEAST_ONCE、EXACTLY_ONCE。
transaction_timeoutInt否600Pulsar 事务超时时间，单位为秒。用于 EXACTLY_ONCE。
pulsar.configMap否-传给 Pulsar 生产者客户端的额外参数。
message.routing.modeEnum否RoundRobinPartition分区 topic 的消息路由模式。可选值：SinglePartition、RoundRobinPartition。
partition_key_fieldsArray否-用于生成 Pulsar 消息 key 的字段。
multi_table_sink_replicaInt否1多表写入时的写入器副本数。
common-optionsConfig否-Sink 插件通用参数，详见 Sink 通用选项。

参数解释​

topic [String]​

Sink 默认写入的 Pulsar topic。

普通单表任务需要配置 topic。多表任务中，如果每行数据带有表标识，Sink 会把这个表标识作为目标 topic；只有当数据没有表标识时才会回退使用 topic。

如果数据没有表标识，并且也没有配置 topic，任务会直接报配置错误。

client.service-url [String]​

Pulsar 客户端服务地址。请使用 Pulsar 协议，例如 pulsar://localhost:6650。

admin.service-url [String]​

Pulsar 管理端 HTTP 地址。

例如：http://my-broker.example.com:8080；如果启用了 TLS，可以使用 https://my-broker.example.com:8443。

auth.plugin-class [String]​

认证插件类名。

auth.params [String]​

认证插件参数。

例如：key1:val1,key2:val2。

format [String]​

数据格式。默认格式为 json。可选 text 和 avro 格式。默认字段分隔符为","。如果自定义分隔符，请添加"field_delimiter"选项。使用 avro 格式时，Avro schema 会从上游数据的 row type 自动推导，无需在 sink 侧单独配置 schema。

field_delimiter [String]​

text 格式使用的字段分隔符。默认值为 ,。

semantics [Enum]​

写入一致性语义。

  • AT_LEAST_ONCE：默认值。作业重启或重试后，消息可能重复。
  • EXACTLY_ONCE：通过 Pulsar 事务写入。Pulsar 集群必须开启事务能力，并且 transaction_timeout 应大于 checkpoint 间隔。
  • NON：不和 checkpoint 协调，直接发送消息。作业重启、重试或网络异常后，数据可能重复或丢失。

transaction_timeout [Int]​

Pulsar 事务超时时间，单位为秒。默认值为 600。

该参数只在 semantics = "EXACTLY_ONCE" 时生效。如果事务没有在超时时间内提交，Pulsar 会自动中止该事务。

pulsar.config [Map]​

Pulsar 生产者客户端的额外参数。这些参数会传给 Pulsar producer。

message.routing.mode [Enum]​

分区 topic 的消息路由模式。

  • SinglePartition：没有消息 key 时，选择一个分区并把所有消息写入该分区；有 key 时，Pulsar 会根据 key 哈希选择分区。
  • RoundRobinPartition：没有消息 key 时，按轮询方式写入不同分区。Pulsar 的轮询发生在批处理延迟边界上，不是逐条消息轮询。

partition_key_fields [Array]​

用于生成 Pulsar 消息 key 的字段。

例如上游数据包含 name 和 age，配置 partition_key_fields = ["name"] 后，会生成类似 {"name":"Jack"} 的 JSON key。所选字段必须存在于上游数据结构中。

multi_table_sink_replica [Int]​

多表写入时的写入器副本数。该配置会对多表 Sink 中的所有目标 topic 生效。

common options​

Sink 插件通用参数，请参考 Sink 通用选项。

任务示例​

写入 FakeSource 数据到 Pulsar​

env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
  FakeSource {
    plugin_output = "fake"
    row.num = 10
    schema = {
      fields {
        c_string = string
        c_int = int
        c_bigint = bigint
        c_double = double
        c_timestamp = timestamp
      }
    }
  }
}

sink {
  Pulsar {
    topic = "topic_test"
    client.service-url = "pulsar://localhost:6650"
    admin.service-url = "http://localhost:8080"
    format = json
    pulsar.config = {
      sendTimeoutMs = 30000
    }
  }
}

使用消息 Key 写入​

sink {
  Pulsar {
    topic = "orders"
    client.service-url = "pulsar://localhost:6650"
    admin.service-url = "http://localhost:8080"
    partition_key_fields = ["order_id"]
    message.routing.mode = "SinglePartition"
    format = json
  }
}

精确一次写入​

env {
  parallelism = 1
  job.mode = "STREAMING"
  checkpoint.interval = 10000
}

sink {
  Pulsar {
    topic = "orders"
    client.service-url = "pulsar://localhost:6650"
    admin.service-url = "http://localhost:8080"
    semantics = "EXACTLY_ONCE"
    transaction_timeout = 600
    format = json
  }
}

多表写入​

当上游数据带有表标识时，Sink 可以把每行数据路由到同名 topic。此时可以不配置 topic。

sink {
  Pulsar {
    client.service-url = "pulsar://localhost:6650"
    admin.service-url = "http://localhost:8080"
    multi_table_sink_replica = 2
    format = json
  }
}

变更日志​

Change Log
ChangeCommitVersion
[Improve][API] Optimize the enumerator API semantics and reduce lock calls at the connector level (#9671)https://github.com/apache/seatunnel/commit/9212a771402.3.12
[improve] pulsar options (#9180)https://github.com/apache/seatunnel/commit/26a2160c802.3.12
[Feature][Checkpoint] Add check script for source/sink state class serialVersionUID missing (#9118)https://github.com/apache/seatunnel/commit/4f5adeb1c72.3.11
[Improve] restruct connector common options (#8634)https://github.com/apache/seatunnel/commit/f3499a6eeb2.3.10
[Improve][dist]add shade check rule (#8136)https://github.com/apache/seatunnel/commit/51ef8000162.3.9
[Feature][Restapi] Allow metrics information to be associated to logical plan nodes (#7786)https://github.com/apache/seatunnel/commit/6b7c53d03c2.3.9
[Improve][API] Make sure the table name in TablePath not be null (#7252)https://github.com/apache/seatunnel/commit/764d8b0bc82.3.7
[Feature][Kafka] Support multi-table source read (#5992)https://github.com/apache/seatunnel/commit/60104602d12.3.6
[PulsarSource]Improve pulsar throughput performance. (#6234)https://github.com/apache/seatunnel/commit/37461f4f3e2.3.4
[Feature][Connector-v2][PulsarSink]Add Pulsar Sink Connector. (#4382)https://github.com/apache/seatunnel/commit/543d2c50862.3.4
[Chore] Remove useless DeserializationFormatFactory and its implement (#5880)https://github.com/apache/seatunnel/commit/f0511544ff2.3.4
fix: update IDENTIFIER = Pulsar for pulsar-datasource on project:seatunnel-web (#5852)https://github.com/apache/seatunnel/commit/3b6de3743e2.3.4
[Improve][Common] Introduce new error define rule (#5793)https://github.com/apache/seatunnel/commit/9d1b2582b22.3.4
Support config column/primaryKey/constraintKey in schema (#5564)https://github.com/apache/seatunnel/commit/eac76b4e502.3.4
[Improve][CheckStyle] Remove useless 'SuppressWarnings' annotation of checkstyle. (#5260)https://github.com/apache/seatunnel/commit/51c0d709ba2.3.4
[Hotfix] Fix com.google.common.base.Preconditions to seatunnel shade one (#5284)https://github.com/apache/seatunnel/commit/ed5eadcf732.3.3
[Feature][Json-format] support read format for pulsar (#4111)https://github.com/apache/seatunnel/commit/7d61ae93e72.3.2
[hotfix][pulsar] Fix the bug that can't consume messages all the time. (#4125)https://github.com/apache/seatunnel/commit/a6705cc5bf2.3.2
[Feature] add cdc multiple table support & fix zeta bughttps://github.com/apache/seatunnel/commit/533ff2c2fa2.3.1
[hotfix][pulsar] PulsarSource consumer ack exception. (#4237)https://github.com/apache/seatunnel/commit/9725d675da2.3.1
Merge branch 'dev' into merge/cdchttps://github.com/apache/seatunnel/commit/4324ee19122.3.1
[Improve][Project] Code format with spotless plugin.https://github.com/apache/seatunnel/commit/423b5830382.3.1
[Improve][Connector-v2][Pulsar] Set the name of the pulsar consumption thread. (#4182)https://github.com/apache/seatunnel/commit/e567203f7d2.3.1
[improve][api] Refactoring schema parse (#4157)https://github.com/apache/seatunnel/commit/b2f573a13e2.3.1
[Improve][build] Give the maven module a human readable name (#4114)https://github.com/apache/seatunnel/commit/d7cd6010512.3.1
[Improve][Project] Code format with spotless plugin. (#4101)https://github.com/apache/seatunnel/commit/a2ab1665612.3.1
[Bug][Connector-v2][PulsarSource]Fix pulsar option topic-pattern bug. (#3989)https://github.com/apache/seatunnel/commit/aee2c580ea2.3.1
[Feature][Connector] add get source method to all source connector (#3846)https://github.com/apache/seatunnel/commit/417178fb842.3.1
[Feature][API & Connector & Doc] add parallelism and column projection interface (#3829)https://github.com/apache/seatunnel/commit/b9164b8ba12.3.1
[Improve][Connector-V2][Pulsar] Unified exception for Pulsar source &… (#3590)https://github.com/apache/seatunnel/commit/4fe93234192.3.0
[Hotfix][OptionRule] Fix option rule about all connectors (#3592)https://github.com/apache/seatunnel/commit/226dc6a1192.3.0
[Hotfix][Connector-V2][Pulsar] fix conditional options (#3504)https://github.com/apache/seatunnel/commit/0066affacf2.3.0
[Feature][Connector][pulsar] expose configurable options in Pulsar (#3341)https://github.com/apache/seatunnel/commit/200faa7c292.3.0
[Connector][Dependency] Add Miss Dependency Cassandra And Change Kudu Plugin Name (#3432)https://github.com/apache/seatunnel/commit/6ac6a0a0cd2.3.0
[chore] fix pulsar consumer comment error (#3356)https://github.com/apache/seatunnel/commit/91e632c5262.3.0
[Connector-V2][ElasticSearch] Add ElasticSearch Source/Sink Factory (#3325)https://github.com/apache/seatunnel/commit/38254e3f262.3.0
[hotfix][connector][pulsar] Fix not being able to mark #noMoreNewSplits when restoring (#2945)https://github.com/apache/seatunnel/commit/5ad69076b32.3.0-beta
Move Handover to common module (#2877)https://github.com/apache/seatunnel/commit/d94a874bcb2.3.0-beta
[hotfix][connector-v2] fix pulsar source exceptions (#2820)https://github.com/apache/seatunnel/commit/8ff0ba70152.2.0-beta
[#2606]Dependency management split (#2630)https://github.com/apache/seatunnel/commit/fc047be69b2.2.0-beta
[SeaTunnel]Simply seatunnel package pipeline. (#2563)https://github.com/apache/seatunnel/commit/9d88b6221a2.2.0-beta
[Improve][Connector-V2] Pulsar support user-defined schema (#2436)https://github.com/apache/seatunnel/commit/16cabe6a352.2.0-beta
[improve][UT] Upgrade junit to 5.+ (#2305)https://github.com/apache/seatunnel/commit/362319ff3e2.2.0-beta
StateT of SeaTunnelSource should extend Serializable (#2214)https://github.com/apache/seatunnel/commit/8c426ef8502.2.0-beta
[doc][connector-v2] pulsar source options doc (#2128)https://github.com/apache/seatunnel/commit/59ce8a2b322.2.0-beta
[api-draft][Optimize] Optimize module name (#2062)https://github.com/apache/seatunnel/commit/f79e3112b12.2.0-beta