PublishKafka 2025.5.31.15¶

捆绑包¶

com.snowflake.openflow.runtime | runtime-kafka-nar

描述¶

使用 Kafka Prod API 将 FlowFile 内容作为消息或单个记录发送给 Apache Kafka。要发送的消息可以是个别 FlowFiles，可以使用用户指定的分隔符（例如换行符）进行分隔，也可以是可由已配置的记录读取器读取的面向记录的数据。用于提取消息的补充 NiFi 处理器是 ConsumeKafka。

标签¶

apache、avro、csv、json、kafka、logs、message、openflow、pubsub、put、record、send

输入要求¶

REQUIRED

支持敏感的动态属性¶

false

属性¶

属性	描述
失败策略	指定处理器在无法将数据发布到 Kafka 时如何处理 FlowFile
FlowFile 属性标头模式	与所有 FlowFile 属性名称匹配的正则表达式。任何名称与该模式匹配的属性都将作为标头添加到 Kafka 消息中。如果未指定，则不会将任何 FlowFile 属性添加为标头。
标头编码	对于作为 Kafka 记录标头添加的任何属性，此属性表示用于序列化标头的字符编码。
Kafka 连接服务	提供与 Kafka Broker 的连接以发布 Kafka 记录
Kafka 键	用于发送消息的键。如果未指定，则使用 FlowFile“kafka.key”属性作为消息键（如果存在）。请注意，设置 Kafka 键并同时进行分界可能会导致使用相同的键生成许多 Kafka 消息。这通常并不是问题，因为 Kafka 不强制或假定消息和键的唯一性。尽管如此，同时设置分界符和 Kafka 键仍存在 Kafka 数据丢失的风险。在 Kafka 上执行主题压缩期间，将根据此键对消息进行重复数据消除。
Kafka 键属性编码	发出的 FlowFiles 具有名为“kafka.key”的属性。此属性规定了应如何编码该属性的值。
消息分隔符	指定用于在单个 FlowFile 中划分多条消息的字符串（解析为 UTF-8）。如果未指定，则将 FlowFile 的全部内容作为单条消息使用。如果已指定，则 FlowFile 的内容将按此分隔符拆分，并将每个部分作为单独的 Kafka 消息发送。要输入特殊字符，例如“换行”，请根据 OS 使用 CTRL+Enter 或 Shift+Enter。
消息键字段	输入记录中应用作 Kafka 消息键的字段的名称。
发布策略	用于将传入 FlowFile 记录发布到 Kafka 的格式。
记录键写入器	用于传出 FlowFiles 的记录键写入器
记录元数据策略	指定记录的元数据（主题和分区）是应来自记录的元数据字段，还是应来自已配置的主题名称和分区/分区器类属性
记录读取器	用于传入 FlowFiles 的记录读取器
记录写入器	用于在将数据发送到 Kafka 之前对数据进行序列化的记录写入器
主题名称	处理器向其发布 Kafka 记录的 Kafka 主题的名称
事务 ID 前缀	指定将作为生成的 UUID 的 KafkaProducer 配置 transactional.id，并将以配置的字符串作为前缀。
事务已启用	指定在与 Kafka 通信时是否提供事务担保。如果向 Kafka 发送数据时出现问题，并且此属性设置为 false，则已经发送到 Kafka 的消息将继续发送给使用者。如果将其设置为 true，则 Kafka 事务将被回滚，此时使用者将无法获得那些消息。将此设置为 true 需要将 [投递保证] 属性设置为 [保证复制投递]。
cks	指定保证消息已发送到 Kafka 的要求。对应于 Kafka 客户端 acks 属性。
compression.type	为发送到 Kafka 的记录指定压缩策略。对应于 Kafka 客户端 compression.type 属性。
max.request.size	请求的最大大小（以字节为单位）。对应于 Kafka 客户端 max.request.size 属性。
分区	为记录指定 Kafka 分区目标。
partitioner.class	指定使用哪个类来计算消息的分区 id。对应于 Kafka 客户端 partitioner.class 属性。

关系¶

名称	描述
failure	无法发送到 Kafka 的任何 FlowFile 都将路由到此关系
success	所有内容均已发送到 Kafka 的 FlowFiles。

写入属性¶

名称	描述
msg.count	为此 FlowFile 发送到 Kafka 的消息数。此属性仅添加到路由到 success 的 FlowFiles。

另请参阅¶

com.snowflake.openflow.runtime.processors.kafka.ConsumeKafka