草庐IT

java - Apache Kafka 根据消息的值对窗口消息进行排序

coder 2024-03-31 原文

我正在尝试找到一种方法来重新排序主题分区内的消息并将排序后的消息发送到新主题。

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

例如:

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

此外，我们为每条消息添加一些消息 key ，以将它们发送到相应的分区。

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

例如，我们可以向主题传递一条具有更大 {system-timestamp} 值的消息。

我研究了 Kafka Stream API 并找到了一些关于消息窗口和聚合的示例:

Properties streamsConfiguration = new Properties();
        streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "stream-sorter");
        streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        streamsConfiguration.put(StreamsConfig.ZOOKEEPER_CONNECT_CONFIG, "localhost:2181");
        streamsConfiguration.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        streamsConfiguration.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());

 KStreamBuilder builder = new KStreamBuilder();
 KStream<String, String> stream = builder.stream("events");
 KGroupedStream<String>, String> groupedStream = stream.groupByKey();//grouped events within partion.

    /* commented since I think that I don't need any aggregation, but I guess without aggregation I can't use time windowing.
KTable<Windowed<String>, String> windowedEvents = stream.groupByKey().aggregate(
                () -> "",  // initial value
                (aggKey, value, aggregate) -> aggregate + "",   // aggregating value
                TimeWindows.of(1000), // intervals in milliseconds
                Serdes.String(), // serde for aggregated value
                "test-store"
        );*/

但是接下来我应该如何处理这个分组流？我没有看到任何可用的“sort() (e1,e2) -> e1.compareTo(e2)”方法，也可以将窗口应用于 aggregation(), reduce() ,count() ，但我认为我不需要任何消息数据操作。

如何在 1 分钟窗口内重新排序消息并将它们发送到另一个主题？

最佳答案

这是一个大纲:

创建一个处理器实现:

  • 在 process() 方法中，对于每条消息:

    • 从消息值中读取时间戳
    • 使用 (timestamp, message-key) 对作为键，使用消息值作为值插入到 KeyValueStore 中。注意，这也提供重复数据删除。您需要提供一个自定义 Serde 来序列化 key ，以便时间戳排在第一位，按字节排列，以便范围查询按时间戳排在第一位。

  • 在 punctuate() 方法中:

    • 使用从 0 到时间戳的范围提取读取存储 - 60'000(=1 分钟)
    • 使用 context.forward() 按顺序发送获取的消息并将它们从存储中删除

这种方法的问题在于，如果没有新消息到达以提前“流时间”，则不会触发 punctuate()。如果这对您的情况有风险，您可以创建一个外部调度程序，它会定期向主题的每个(!)分区发送“滴答”消息，您的处理器应该忽略这些消息，但它们会导致标点符号在不存在的情况下触发“真实”的消息。 KIP-138 将通过添加对系统时间标点符号的明确支持来解决此限制: https://cwiki.apache.org/confluence/display/KAFKA/KIP-138%3A+Change+punctuate+semantics

关于java - Apache Kafka 根据消息的值对窗口消息进行排序，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/43939534/

ApacheKafkastreamsConfigurationStringsectionjavastreamapache-kafkamessaging

有关java - Apache Kafka 根据消息的值对窗口消息进行排序的更多相关文章

  1. ruby-on-rails - 使用 Ruby on Rails 进行自动化测试 - 最佳实践 - 2

    很好奇，就使用ruby​​onrails自动化单元测试而言，你们正在做什么？您是否创建了一个脚本来在cron中运行rake作业并将结果邮寄给您？git中的预提交Hook？只是手动调用？我完全理解测试，但想知道在错误发生之前捕获错误的最佳实践是什么。让我们理所当然地认为测试本身是完美无缺的，并且可以正常工作。下一步是什么以确保他们在正确的时间将可能有害的结果传达给您？ 最佳答案 不确定您到底想听什么，但是有几个级别的自动代码库控制:在处理某项功能时，您可以使用类似autotest的内容获得关于哪些有效，哪些无效的即时反馈。要确保您的提

  2. ruby-on-rails - 按天对 Mongoid 对象进行分组 - 2

    在控制台中反复尝试之后，我想到了这种方法，可以按发生日期对类似activerecord的(Mongoid)对象进行分组。我不确定这是完成此任务的最佳方法，但它确实有效。有没有人有更好的建议，或者这是一个很好的方法？#eventsisanarrayofactiverecord-likeobjectsthatincludeatimeattributeevents.map{|event|#converteventsarrayintoanarrayofhasheswiththedayofthemonthandtheevent{:number=>event.time.day,:event=>ev

  3. ruby - 使用 C 扩展开发 ruby​​gem 时，如何使用 Rspec 在本地进行测试？ - 2

    我正在编写一个包含C扩展的gem。通常当我写一个gem时，我会遵循TDD的过程，我会写一个失败的规范，然后处理代码直到它通过，等等......在“ext/mygem/mygem.c”中我的C扩展和在gemspec的“扩展”中配置的有效extconf.rb，如何运行我的规范并仍然加载我的C扩展？当我更改C代码时，我需要采取哪些步骤来重新编译代码？这可能是个愚蠢的问题，但是从我的gem的开发源代码树中输入“bundleinstall”不会构建任何native扩展。当我手动运行rubyext/mygem/extconf.rb时，我确实得到了一个Makefile(在整个项目的根目录中)，然后当

  4. java - 等价于 Java 中的 Ruby Hash - 2

    我真的很习惯使用Ruby编写以下代码:my_hash={}my_hash['test']=1Java中对应的数据结构是什么？ 最佳答案 HashMapmap=newHashMap();map.put("test",1);我假设？ 关于java-等价于Java中的RubyHash，我们在StackOverflow上找到一个类似的问题： https://stackoverflow.com/questions/22737685/

  5. ruby - 如何进行排列以有效地定制输出 - 2

    这是一道面试题，我没有答对，但还是很好奇怎么解。你有N个人的大家庭，分别是1,2,3,...,N岁。你想给你的大家庭拍张照片。所有的家庭成员都排成一排。“我是家里的friend，建议家庭成员安排如下:”1岁的家庭成员坐在这一排的最左边。每两个坐在一起的家庭成员的年龄相差不得超过2岁。输入:整数N，1≤N≤55。输出:摄影师可以拍摄的照片数量。示例->输入:4，输出:4符合条件的数组:[1,2,3,4][1,2,4,3][1,3,2,4][1,3,4,2]另一个例子:输入:5输出:6符合条件的数组:[1,2,3,4,5][1,2,3,5,4][1,2,4,3,5][1,2,4,5,3][

  6. ruby - 如果指定键的值在数组中相同，如何合并哈希 - 2

    我有一个这样的哈希数组:[{:foo=>2,:date=>Sat,01Sep2014},{:foo2=>2,:date=>Sat,02Sep2014},{:foo3=>3,:date=>Sat,01Sep2014},{:foo4=>4,:date=>Sat,03Sep2014},{:foo5=>5,:date=>Sat,02Sep2014}]如果:date相同，我想合并哈希值。我对上面数组的期望是:[{:foo=>2,:foo3=>3,:date=>Sat,01Sep2014},{:foo2=>2,:foo5=>5:date=>Sat,02Sep2014},{:foo4=>4,:dat

  7. ruby - 如何根据特征实现 FactoryGirl 的条件行为 - 2

    我有一个用户工厂。我希望默认情况下确认用户。但是鉴于unconfirmed特征，我不希望它们被确认。虽然我有一个基于实现细节而不是抽象的工作实现，但我想知道如何正确地做到这一点。factory:userdoafter(:create)do|user,evaluator|#unwantedimplementationdetailshereunlessFactoryGirl.factories[:user].defined_traits.map(&:name).include?(:unconfirmed)user.confirm!endendtrait:unconfirmeddoenden

  8. ruby - 检查字符串是否包含散列中的任何键并返回它包含的键的值 - 2

    我有一个包含多个键的散列和一个字符串，该字符串不包含散列中的任何键或包含一个键。h={"k1"=>"v1","k2"=>"v2","k3"=>"v3"}s="thisisanexamplestringthatmightoccurwithakeysomewhereinthestringk1(withspecialcharacterslike(^&*$#@!^&&*))"检查s是否包含h中的任何键的最佳方法是什么，如果包含，则返回它包含的键的值？例如，对于上面的h和s的例子，输出应该是v1。编辑:只有字符串是用户定义的。哈希将始终相同。 最佳答案

  9. ruby - 即使失败也继续进行多主机测试 - 2

    我已经构建了一些serverspec代码来在多个主机上运行一组测试。问题是当任何测试失败时，测试会在当前主机停止。即使测试失败，我也希望它继续在所有主机上运行。Rakefile:namespace:specdotask:all=>hosts.map{|h|'spec:'+h.split('.')[0]}hosts.eachdo|host|begindesc"Runserverspecto#{host}"RSpec::Core::RakeTask.new(host)do|t|ENV['TARGET_HOST']=hostt.pattern="spec/cfengine3/*_spec.r

  10. ruby-on-rails - 如何在 Rails View 上显示错误消息？ - 2

    我是rails的新手，想在form字段上应用验证。myviewsnew.html.erb.....模拟.rbclassSimulation{:in=>1..25,:message=>'Therowmustbebetween1and25'}end模拟Controller.rbclassSimulationsController我想检查模型类中row字段的整数范围，如果不在范围内则返回错误信息。我可以检查上面代码的范围，但无法返回错误消息提前致谢 最佳答案 关键是您使用的是模型表单，一种显示ActiveRecord模型实例属性的表单。c

随机推荐