1

我正在尝试找到一种方法来重新排序我的 Kafka 消息并将有序消息发送到使用Apache Beam和Google DataFlow的新主题。

我有发送以下格式的字符串消息的 Kafka 发布者： {system_timestamp}-{event_name}?{parameters}

例如：

1494002667893-client.message?chatName=1c&messageBody=hello
1494002656558-chat.started?chatName=1c&chatPatricipants=3

我想做的是根据消息的{system-timestamp}部分并在 5 秒的窗口内重新排序事件，因为我们的发布者不保证消息将按照{system-timestamp}值发送。

我编写了一个模拟排序器函数，对从 Kafka 接收到的事件进行排序（使用KafkaIO源）：

static class SortEventsFunc extends DoFn<KV<String, Iterable<String>>, KV<String, Iterable<String>>> {

   @ProcessElement
   public void processElement(ProcessContext c) {
       KV<String, Iterable<String>> element = c.element();

       System.out.println("");
       System.out.print("key: " + element.getKey() + ";");

       Iterator<String> it = element.getValue().iterator();
       List<String> list = new ArrayList<>();
       while (it.hasNext()) {
           String val = it.next();
           System.out.print("value: " + val);
           list.add(val);
       }
       Collections.sort(list, Comparator.naturalOrder());
       c.output(KV.of(element.getKey(), list));
   }
 }

public static void main(String[] args) {
    PipelineOptions options = PipelineOptionsFactory.create();

    DirectOptions directOptions = options.as(DirectOptions.class);
    directOptions.setRunner(DirectRunner.class);

    // Create the Pipeline object with the options we defined above.
    Pipeline pipeline = Pipeline.create(options);
    pipeline
        // read from Kafka
        .apply(KafkaIO.<String,String>read()
            .withBootstrapServers("localhost:9092")
            .withTopics(new ArrayList<>((Arrays.asList("events"))))
            .withKeyDeserializer(StringDeserializer.class)
            .withValueDeserializer(StringDeserializer.class)
            .withoutMetadata())
        // apply window
        .apply(Window.<KV<String,String>>into(
                FixedWindows.of(Duration.standardSeconds(5L))))
        // group by key before sorting
        .apply(GroupByKey.<String, String>create()) // return PCollection<KV<String, Iterable<String>>
        // sort events
        .apply(ParDo.of(new SortEventsFunc()))
        //combine KV<String, Iterable<String>> input to KafkaIO acceptable KV<String, String> format
        .apply(Combine.perKey()) //:TODO somehow convert KV<String, Iterable<String>> to KV<String, String>
        // write ordered events to Kafka
        .apply(KafkaIO.<String, String>write()
                .withBootstrapServers("localhost:9092")
                .withTopic("events-sorted")
                .withKeySerializer(StringSerializer.class)
                .withValueSerializer(StringSerializer.class)
            );
    pipeline.run();
}

因此，我使用GroupByKey.<String, String>create()转换对消息进行了分组，在 sortrin 事件之后，我需要以某种方式将它们从KafkaIO值转换KV<String, Iterable<String>>为接受。KV<String, String> or KV<Void, String>所以我想做的就是忽略通过分组转换键创建的，只需将每个值作为单独的消息传递给 KafkaIO writer。

我探索了Combine#perKey转换，但它接受只能将所有值组合到一个字符串（带有一些分隔符）的SerializableFunctionKafkaIO#read() ，因此我只将一个值作为一个连接字符串而不是每个值（由 读取）传递给 KafkaIO 写入器。

4

1 回答 1

3

其实很简单！这里的诀窍是你可以c.output在方法内部调用任意多次@ProcessElement。

因此，在您的情况下，只需定义一个DoFn<KV<String, Iterable<String>>, KV<String, String>>，遍历c.element().getValue()集合，然后调用c.output它们中的每一个。

于 2017-05-29T22:56:57.987 回答