pipelinedb消费kafka主题答案

【问题标题】：pipelinedb consuming kafka topicpipelinedb消费kafka主题
【发布时间】：2017-04-25 17:35:55
【问题描述】：

当我在管道中创建一个流并开始消费一个主题时，它没有出现在我的主题的活动 consmidores 中，我该怎么做才能让它出现在那里？

我是如何创建流的

CREATE STREAM pipeline_kafka.summary_device_stream(pack json);
SELECT pipeline_kafka.consume_begin('topicSummaryDevice', 'pipeline_kafka.summary_device_stream', format := 'json');

【问题讨论】：

标签： apache-kafka pipelinedb

【解决方案1】：

我不确定您所附屏幕截图中该值的确切来源，但我猜它仅适用于在 Kafka 中存储偏移量的消费者。

您可以通过将group_id 参数传递给消费者命令，使用pipeline_kafka 在Kafka 中存储偏移量，如下所示：

SELECT pipeline_kafka.consume_begin('topicSummaryDevice',
   'pipeline_kafka.summary_device_stream',
    group_id := 'consumer_group_id',
    format := 'json');

【讨论】：

严重错误 :( SQL 错误 [42883]: 错误: 函数 pipeline_kafka.consume_begin(unknown, unknown, group_id => unknown, format => unknown) 不存在 Dica: 没有函数匹配给定的名称和参数类型。您可能需要添加显式类型转换。Posição: 8 org.postgresql.util.PSQLException: ERROR: function pipeline_kafka.consume_begin(unknown, unknown, group_id => unknown, format => unknown) 没有存在 Dica：没有函数匹配给定的名称和参数类型。您可能需要添加显式类型转换。Posição：8
您输入的导致该错误的确切命令是什么？
这个命令...SELECT pipeline_kafka.consume_begin('topicNotificationProcess', 'pipeline_kafka.sensor_event_process_stream', format := 'json', group_id:= 'groupEventProcess'; batchsize := 4330, parallelism := 5, start_offset := -1);
该命令在group_id 参数之后有一个;，而不是,，这实际上甚至不是有效的语法。除此之外，您应该通过检查\dx 的输出来验证是否安装了pipeline_kafka 扩展。