Kafka Streams 表转换

Question

我在 SQL Server 中有一个表，我想流式传输到 Kafka 主题，结构如下：

(UserID, ReportID)

此表将不断更改（添加、插入、无更新记录）

我想把这个变成这样的结构，放到 Elasticsearch 中：

{
  "UserID": 1,
  "Reports": [1, 2, 3, 4, 5, 6]
}

到目前为止，我看到的示例是日志或点击流，它们在我的情况下不起作用。

这种用例可能吗？我总是可以只看UserID 更改和查询数据库，但这似乎很幼稚，并不是最好的方法。

更新

import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.serialization.Serializer;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.*;

import java.util.ArrayList;
import java.util.Properties;

public class MyDemo {
  public static void main(String... args) {
    System.out.println("Hello KTable!");

    final Serde<Long> longSerde = Serdes.Long();

    KStreamBuilder builder = new KStreamBuilder();

    KStream<Long, Long> reportPermission = builder.stream(TOPIC);

    KTable<Long, ArrayList<Long>> result = reportPermission
        .groupByKey()
        .aggregate(
            new Initializer<ArrayList<Long>>() {
              @Override
              public ArrayList<Long> apply() {
                return null;
              }
            },
            new Aggregator<Long, Long, ArrayList<Long>>() {
              @Override
              public ArrayList<Long> apply(Long key, Long value, ArrayList<Long> aggregate) {
                aggregate.add(value);
                return aggregate;
              }
            },
            new Serde<ArrayList<Long>>() {
              @Override
              public void configure(Map<String, ?> configs, boolean isKey) {}

              @Override
              public void close() {}

              @Override
              public Serializer<ArrayList<Long>> serializer() {
                return null;
              }

              @Override
              public Deserializer<ArrayList<Long>> deserializer() {
                return null;
              }
            });

    result.to("report-aggregated-topic");

    KafkaStreams streams = new KafkaStreams(builder, createStreamProperties());
    streams.cleanUp();
    streams.start();

    Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
  }

  private static final String TOPIC = "report-permission";

  private static final Properties createStreamProperties() {
    Properties props = new Properties();

    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "report-permission-app");
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092");

    return props;
  }
}

我实际上陷入了聚合阶段，因为我无法为 ArrayList<Long> 编写合适的 SerDe（还没有足够的技能），lambda 似乎不适用于聚合器 - 它不知道 agg 的类型是什么：

KTable<Long, ArrayList<Long>> sample = builder.stream(TOPIC)
    .groupByKey()
    .aggregate(
        () -> new ArrayList<Long>(),
        (key, val, agg) -> agg.add(val),
        longSerde
    );

Answer 1

您可以使用 Kafka 的 Connect API 将数据从 SQL Server 获取到 Kafka。我不知道 SQL Server 的任何特定连接器，但您可以使用任何基于 JDBC 的通用连接器：https://www.confluent.io/product/connectors/

要处理数据，您可以使用 Kafka 的 Streams API。您可以简单地 aggregate() all 报告每个用户。像这样的：

KTable<UserId, List<Reports>> result =
    builder.stream("topic-name")
           .groupByKey()
           // init a new empty list and
           // `add` the items to the list in the actual aggregation
           .aggregate(...);

result.to("result-topic");

查看文档以了解有关 Streams API 的更多详细信息：https://docs.confluent.io/current/streams/index.html

请注意，您需要确保报告列表不会无限增长。 Kafka 有一些（可配置的）最大消息大小，整个列表将包含在单个消息中。因此，您应该在投入生产之前估计最大消息大小并应用相应的配置 (-> max.message.bytes)。在网页上查看配置：http://kafka.apache.org/documentation/#brokerconfigs

最后，您使用 Connect API 将数据推送到 Elastic Search。有多种不同的连接器可用（我当然会推荐 Confluent 一个）。有关 Connect API 的更多详细信息：https://docs.confluent.io/current/connect/userguide.html

Answer 2

在SQL和Kafka Streams中直接不允许这种方法，但是用例是可能的，可以实现如下：

1) 使用 SOLRJ API 在 SQL 服务器上编写自定义应用程序，每当在 SQL 中执行 DML（插入、更新、删除等）操作时，该应用程序将命中 Solr 实例。 https://wiki.apache.org/solr/Solrj

2) 使用 Solr SQL 数据导入处理程序，只要 SQL 中发生 DML（插入、更新、删除等）操作，SQL Server 就会自动通知 solr。 https://wiki.apache.org/solr/DataImportHandler