Kafka

DolphinDB Server: 2.00.10 及更高版本，支持 x86-64 Linux，ARM Linux。

其中 ARM Linux 版本的 Kafka 插件不支持 SASL 验证，且在生产和消费时不支持 zstd 压缩格式。

1. 在 DolphinDB 客户端中使用 listRemotePlugins 命令查看插件仓库中的插件信息。

   login("admin", "123456")
   listRemotePlugins()

2. 使用 installPlugin 命令完成插件安装。

   installPlugin("kafka")

3. 使用 loadPlugin 命令加载插件。

   loadPlugin("kafka")

语法

kafka::producer(config, [errCallback])

详情

根据指定配置创建一个 Kafka 生产者并返回。注意，目前没有显式的函数用于关闭 producer，如要删除可以将其置空触发析构。Server 的 close 函数无法关闭 producer。

参数

config 字典，表示 Kafka 生产者的配置。字典的 key 为字符串，value 为字符串或布尔值。

producer 的配置一般只需要填写地址就可以进行连接，如果 Kafka 集群启用了安全校验则需要有更多的配置选项。有关 Kafka 配置的详细信息，请参阅 librdkafka/CONFIGURATION.md at master · confluentinc/librdkafka 。

errCallback 可选参数，函数类型，用于在后台发生错误（ERROR）或者警告（WARNING）情况下进行回调。该函数必须有三个入参，都是字符串类型，分别表示 level 错误等级，errorName 错误名称，message 具体错误信息。level 的值只可能为 “ERROR” 或者 “WARNING”。

Kafka 通用配置参数

以下通用参数也可以在 consumer 中使用。

Kafka producer 专属配置参数

示例

创建一个普通连接本地 Kafka 的 producer。

producerCfg = dict(STRING, ANY)
producerCfg["metadata.broker.list"] = "localhost:9092"
handle = kafka::producer(producerCfg)

创建一个带有回调的 producer。

producerCfg = dict(string, any)
producerCfg["bootstrap.servers"] = "192.168.100.3"
share table(1:0, `level`err`reason, [STRING,STRING,STRING]) as produceDest
def producerCallback(mutable dest, level, err, reason) {
    tb = table([level] as level, [err] as err, [reason] as reason)
    dest.append!(tb)
}
producer = kafka::producer(producerCfg, producerCallback{produceDest})

语法

kafka::produce(producer, topic, key, value, marshalType, [partition])

参数

producer Kafka 生产者的句柄。

topic STRING 类型标量，要将数据发送到的 Kafka 的 topic。

key STRING 类型标量，需要发送的 Kafka 消息的 key。Kafka 的分区函数将会根据 key 的值进行分区，填写了相同 key 的消息会被分配到同一个分区中。

value STRING 类型标量，需要发送的 Kafka 消息体对应的数据。

marshalType 可选参数，STRING 类型标量，表示以何种格式序列化数据。具体取值可以为 "JSON", "DOLPHINDB", "PLAIN"，分别表示使用 JSON 格式序列化、DolphinDB 内部格式序列化，以及不进行序列化直接发送。最后一种方式只能发送字符串。

partition 可选参数，整型标量，表示要发送到的 Kafka topic 分区号。如果未填写则 Kafka 集群会将消息均匀地按 key 进行分区分配到 topic 的各个分区下。

详情

以选定的序列化格式，将数据发送到 Kafka 集群对应的 topic 中。具体的使用方式以及与 consumer 的配合见 使用示例。

注意

• JSON 格式如果发送的是表格类型数据，会将一列的数据都写入一个数组中。
• 使用 DolphinDB 序列化格式发送的数据只能使用 Kafka 插件并指定对应的 DOLPHINDB marshalType 进行接收，否则二进制序列化数据将无法被处理。极有可能导致乱码的问题。

语法

kafka::producerFlush(producer);

参数

producer Kafka 生产者的句柄。

详情

将生产者的所有缓存记录发送到 Kafka。通常在发送可能产生堆积时，调用该函数。

语法

kafka::consumer(config, [errCallback])

详情

根据指定配置创建一个 Kafka 消费者，并返回。

注意，目前没有显式的函数用于关闭 consumer，如要删除可以将其置空触发析构。如果该 consumer 被用于 kafka::createSubJob 的话，如果想要关闭该 consumer，还需取消对应的后台任务。server 的 close 函数无法关闭 consumer。

参数

config 字典，表示 Kafka 消费者的配置。字典的 key 是一个字符串，value 是字符串或布尔值。

consumer 的配置中 metadata.broker.list 与 group.id 为必填项，通用配置参数的含义参考 kafka::producer 一节，有关 Kafka 配置的更多信息，请参阅 https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md 。

errCallback 可选参数，函数类型，用于在后台发生错误（ERROR）或者警告（WARNING）情况下进行回调。该函数必须有三个入参，都是字符串类型，分别表示 level 错误等级，errorName 错误名称，message 具体错误信息。level 的值只可能为 “ERROR” 或者 “WARNING”。

kafka consumer 专属配置参数

示例

创建一个普通连接本地 Kafka 的 consumer，指定 group.id 为 group1。

consumerCfg = dict(STRING, ANY)
consumerCfg["group.id"] = "group1"
consumerCfg["metadata.broker.list"] = "localhost:9092";
consumer = kafka::consumer(consumerCfg)

语法

kafka::subscribe(consumer, topics)

参数

consumer Kafka 消费者的句柄。

topics STRING 类型标量或者向量，表示订阅的主题。

详情

订阅一个 Kafka 主题，无返回值。

注意：

1. 在使用插件时，同一消费者组中消费者的数量不要超过 topic 分区数，否则即使有分配了分区的消费者退出，先前没分配到分区的消费者也无法参与到 rebalance 中。
2. 如果订阅了不存在的 topic 或者订阅失败，并不会有返回信息与 log，但可以通过 kafka::getAssignment 函数查看订阅情况。

语法

kafka::unsubscribe(consumer)

参数

consumer 要取消 topic 订阅的 Kafka 消费者句柄。

详情

取消该消费者的所有订阅。

注意，即使是使用 kafka::assign 方式指定的以偏移量订阅，也能被 kafka::unsubscribe 所取消，需要谨慎使用。

语法

kafka::consumerPoll(consumer, [timeout=1000], [marshalType])

参数

consumer Kafka 消费者的句柄。

timeout INT 类型标量，可选。表示请求获取消息的最大等待时间，单位为毫秒。

marshalType STRING 类型标量，可选。表示以何种格式解析数据。具体取值可以为 “JSON”, ”DOLPHINDB”, ”PLAIN”，分别表示使用 JSON 格式解析、DolphinDB 内部格式解析，以及直接返回原始字符串。如果不指定，则会进行检测，选择 JSON 或者 DOLPHINDB 格式进行解析。

详情

从 Kafka 集群中获取订阅的消息，返回一个元组。第一个元素表示出错信息的字符串，若成功获取则为空。 第二个元素是一个元组，其元素包括：主题、分区、键、值和消费者收到数据的时间戳。具体的使用方式以及与 consumer 的配合见 使用示例。

语法

kafka::consumerPollBatch(consumer, batchSize, [timeout=1000], [marshalType])

参数

consumer Kafka 消费者的句柄。

batchSize INT 类型标量，表示要批量消费的条数。

timeout INT 类型标量，可选。表示请求获取消息的最大等待时间，单位为毫秒。

marshalType STRING 类型标量，可选。表示以何种格式解析数据。具体取值可以为 “JSON”, ”DOLPHINDB”, ”PLAIN”，分别表示使用 json 格式解析、DolphinDB 内部格式解析，以及直接返回原始字符串。如果不指定，则会进行检测，基于检测结果，选择 JSON 或者 DOLPHINDB 格式进行解析。

详情

从 Kafka 集群中批量获取订阅的消息，返回一个元组。其中每个元素都是一个元组，其元素包括：主题、分区、键、值和消费者收到数据的时间戳。具体的使用方式以及与 consumer 的配合见 使用示例。

语法

kafka::createSubJob(consumer, table, parser, actionName, [throttle=1.0], [autoCommit], [msgAsTable=false], [batchSize=1], [queueDepth=1000000])

consumer 消费者的句柄。

table 表类型，可以是共享流表、共享内存表、共享键值表、共享索引表或 DFS 表。订阅数据会直接插入到该表中。

parser 可以为一元函数、二元函数或者三元函数，用于处理 Kafka 的订阅数据。返回类型为表，返回结果将输出到 table 参数中：

• 如果指定 msgAsTable 为 false，输入参数均为 STRING 类型，parser 的参数数目可以为 1-3 个，第一个参数为 Kafka 消息的 value，第二个参数为 Kafka 消息的 key，第三个参数为 Kafka 消息的 topic。
• 如果指定 msgAsTable 为 true，parser 的输入参数需要为一个表。

actionName 是一个字符串，表示订阅任务的名称，与已经创建的订阅名称不可重复。

throttle 是一个浮点数，单位为秒，默认值为 1.0。表示继上次 parser 处理消息之后，若未达到 batchSize 条件，多久后再次处理消息。仅当 msgAsTable 为 true 时有效。

autoCommit 是一个布尔值，无默认值。表示是否在 parser 处理消息完成、插入到 table 数据表之后进行一次 Kafka 的 commit。

• 如果 consumer 对应的 comsumer 不会自动 commit，开启 autoCommit 可以防止因为系统崩溃而在插件端丢失 Kafka 订阅数据。因为没有进入 table 的数据没有被 commit，会随着系统恢复而被插件重新接收。
• 如果传入的 Kafka 的 config 中指定了 enable.auto.commit 为 false，且 autoCommit 没有指定为 false，则也会在消息处理后进行 commit 操作。

msgAsTable 是一个布尔值，默认值是 false。表示订阅的数据是否为表。

• 如果为 false，输入到 parser 的数据视 parser 参数个数而定，参数数目可以为 1-3 个，第一个参数为 msg 的 value，第二个参数为 msg 的 key，第三个参数为 msg 的 topic。
• 如果为 true，输入到 parser 的数据将为一个三列 STRING 类型的表，三列名称分别为 value、key、topic。

batchSize 是一个整数，默认值为 0。若为正数，表示未处理消息的数量达到 batchSize 时，parser 才会处理消息。若未指定或为非正数，每一批次的消息到达之后，handler 会立即处理。仅当 msgAsTable 为 true 时有效。

queueDepth 是一个整数，大于 0，默认值为 1000000。表示 Kafka 后台接收数据队列的深度。

详情

返回一个 subJob connection handle。具体的使用方式见 使用示例。

语法

kafka::getJobStat()

详情

返回一个表，包含通过 kafka::createSubJob 所创建的后台订阅任务的信息。

语法

kafka::cancelSubJob(handle|actionName)

参数

handle|actionName 可以为函数 kafka::createSubJob 的返回值，也可以为创建任务时所指定的 actionName。

详情

取消对应的 Kafka subJob 后台任务。在取消后 kafka::getJobStat 也将无法查询到对应的记录。

注意，为保持兼容性，函数也可以传入 kafka::getJobStat 中对应记录的 subscriptionId，如果 actionName 与 subscriptionId 同名，则优先取消 actionName 对应的后台任务。

语法

kafka::getSubJobConsumer(handle|actionName)

参数

handle|actionName 可以为函数 kafka::createSubJob 的返回值，也可以为创建任务时所指定的 actionName。

详情

每个 kafka::createSubJob 创建的后台任务都会有一个对应的 consumer handle，如果需要获取特定后台订阅中的 consumer 进行特殊处理，可以使用此接口。

语法

kafka::commit(consumer, [topics], [partitions], [offsets])

参数

consumer Kafka 消费者的句柄。

topics 可选，字符串标量或向量，表示订阅的主题。

partitions 可选，整型标量或向量，表示对应主题的分区。

offsets 可选，整型标量或向量，表示每个主题的偏移量。

topics, partitions, offsets 须保证形式相同，长度相同，其中的元素一一对应。

详情

将 consumer 最新消费数据的偏移量同步给 Kafka 集群。

如果指定了 topic、partition、offset，会以指定的信息进行 commit。

如果没有指定，则会 commit 传入的 consumer 所订阅的所有分区的最新消费 offset。

语法

kafka::assign(consumer, topics, partitions, offsets)

参数

consumer Kafka 消费者的句柄。

topics 字符串标量或向量，表示订阅的主题。

partitions 整型标量或向量，表示每个主题的分区。

offsets 整型标量或向量，表示每个主题的偏移量。

topics, partitions, offsets 须保证形式相同，长度相同，其中的元素一一对应。

详情

与 kafka::subscribe 不同，该函数为消费者手动指定特定的主题、分区和偏移量。

注意

• 当执行该函数后，消费者组将不会对这个 consumer 产生约束，即可以有多个 consumer 通过 assign 的方式消费同一个主题的同一个分区的数据。
• 在使用时，尽量避免对同一个消费者同时操作 kafka::subscribe 与 kafka::assign 操作，也尽量避免操作将以上函数执行多次，可能会出现一些未定义行为。

示例

指定 consumer 消费 test 主题，分区 1，偏移量为 3000 的消息。

kafka::assign(consumer, "test", 1, 3000)

语法

kafka::unassign(consumer)

参数

consumer Kafka 消费者的句柄。

详情

取消 consumer 订阅的主题，对通过 kafka::subscribe 与 kafka::assign 执行的订阅都可以生效。

语法

kafka::getAssignment(consumer)

参数

consumer Kafka 消费者的句柄。

详情

获取 consumer 对应的 offset assign 情况。返回一个 table，包含 3 列，topic 代表主题，partition 代表分区号，offset 代表偏移量。

注意，即便对应的 consumer 没有调用 assign 函数，只调用了 subscribe 函数，也能够得到具体的返回。

语法

kafka::getOffsetInfo(consumer, topics, partitions)

参数

consumer Kafka 消费者的句柄。

topics 字符串标量或向量，表示订阅的主题。

partitions 整型标量或向量，表示每个主题的分区。

topics, partitions 须保证形式相同，长度相同，其中的元素一一对应。

详情

获取 consumer 对应的 topic 和 partition 的偏移量信息。注意，consumer 需要已经 subscribe 或者 assign 该 topic、partition，否则无法查出有效的信息。

返回一个 table，字段内容如下：

示例

kafka::getAssignment 与 kafka::getOffsetInfo 的使用常常引起疑问，这里举一个例子说明用法和各个字段的含义。

首先发送 10 条消息到 offsetTopic 这个新的 topic 中。

producerCfg = dict(string, any);
producerCfg["bootstrap.servers"] = "localhost"
producer = kafka::producer(producerCfg)
for (i in 0..9) {
    kafka::produce(producer, "offsetTopic", "0", "message", "PLAIN")
}

然后创建 consumer，消费两次。

consumerCfg = dict(string, any);
consumerCfg["metadata.broker.list"] = "localhost"
consumerCfg["enable.auto.commit"] = "false"
consumerCfg["group.id"] = "test"
consumer = kafka::consumer(consumerCfg)
kafka::assign(consumer, "offsetTopic", 0, 3)
kafka::consumerPoll(consumer,,"PLAIN")
kafka::consumerPoll(consumer,,"PLAIN")

此时执行 kafka::getAssignment 能看到订阅了 partition 0，offset 为 -1000 说明未指定具体的订阅 offset。

kafka::getAssignment(consumer)

而执行 kafka::getOffsetInfo 会看到 offsetTopic 这个主题的 partition 0 的各类 offset 数据。其中 minOffset 为 0，因为发送了 10 条消息所以 maxOffset 为 10，当前 consumer 消费了 2 条数据，所以 offsetPosition 为 2。由于建立 consumer 时指定了不自动 commit，所以 offsetCommitted 为无意义值即尚未 commit 过。

kafka::getOffsetInfo(consumer, "offsetTopic", 0)

如果执行一遍 kafka::commit，再查询偏移量相关信息，则会发现 offsetCommitted 字段变成了 2，即当前消费并 commit 的 offset。

kafka::commit(consumer)
kafka::getOffsetInfo(consumer, "offsetTopic", 0)

语法

kafka::getMemId(consumer)

参数

consumer Kafka 消费者的句柄。

详情

返回一个字符串。每个 consumer 成员会有一个对应的 ID，可以通过该函数获取传入 consumer 的 ID。

示例

consumerCfg = dict(STRING, ANY)
consumerCfg["group.id"] = string(now())
consumerCfg["metadata.broker.list"] = "localhost:9092";
consumer = kafka::consumer(consumerCfg)
kafka::subscribe(consumer, "test");
kafka::consumerPoll(consumer); // 因为如果不进行数据的拉取，可能在 Kafka 集群端不会被分配对应 ID
kafka::getMemId(consumer)
// output: 'rdkafka-d9eded44-358f-49fc-be01-cc099b121d59'

语法

kafka::getMetadata(handle|conf)

参数

handle|conf 可以为已经建立的 producer、consumer，或者用于建立连接的 dict，或者 Kafka cluster 的 IP PORT。

详情

用于获取对应 Kafka 集群的元数据信息。返回一个字典，包括以下键值对：

示例

直接查询 localhost Kafka 的元数据。

kafka::getMetadata("localhost:9092")

查询 consumer handle 对应 Kafka 集群的元数据。

kafka::getMetadata(consumer)

输出示例：

brokers->
  id host           port
  -- -------------- ----
  2  192.168.100.45 9092
  1  192.168.100.44 9092
  0  192.168.100.43 9092
  
consumerGroup->
  test->
    state->Stable
    protocolType->consumer
    protocol->range
    error->Success
    members->
      rdkafka-0b501728-5344-4f2f-a450-bc056f6e3400->
        clientID->rdkafka
        clientHost->/192.168.0.38
        memberAssignmentVersion->0
        partitions->
          topic partition offset
          ----- --------- ------
          test  0         -1001 
          test  1         -1001 
          test  2         -1001 

topics->
  test->
    id error   leader replicas
    -- ------- ------ --------
    0  Success 0      [0]     
    1  Success 1      [1]     
    2  Success 2      [2]