kafka.to() 函数

kafka.to() 函数将数据发送到 Apache Kafka brokers。

函数类型签名

(
    <-tables: stream[A],
    brokers: [string],
    topic: string,
    ?balancer: string,
    ?name: string,
    ?nameColumn: string,
    ?tagColumns: [string],
    ?timeColumn: string,
    ?valueColumns: [string],
) => stream[A] where A: Record

参数

brokers

(必需) 要向其发送数据的 Kafka brokers 列表。

topic

(必需) 要向其发送数据的 Kafka topic。

balancer

Kafka 负载均衡策略。默认为 hash。

负载均衡策略决定了消息如何路由到 Kafka 集群上可用的分区。以下策略可用：

• hash：使用组键的哈希值来确定将消息路由到哪个 Kafka 分区。这确保了来自表行的消息被路由到同一分区。
• round-robin：在所有可用分区之间平均分配消息。
• least-bytes：将消息路由到接收数据量最少的分区。

name

Kafka 指标名称。默认为 nameColumn 的值。

nameColumn

用作 Kafka 指标名称的列。默认为 _measurement。

timeColumn

时间列。默认为 _time。

tagColumns

输入数据中的标签列列表。

valueColumns

输入数据中的值列列表。默认为 ["_value"]。

tables

输入数据。默认为管道转发数据 (<-)。

示例

发送数据到 Kafka

import "kafka"
import "sampledata"

sampledata.int()
    |> kafka.to(
        brokers: ["http://127.0.0.1:9092"],
        topic: "example-topic",
        name: "example-metric-name",
        tagColumns: ["tag"],
    )