- html - 出于某种原因,IE8 对我的 Sass 文件中继承的 html5 CSS 不友好?
- JMeter 在响应断言中使用 span 标签的问题
- html - 在 :hover and :active? 上具有不同效果的 CSS 动画
- html - 相对于居中的 html 内容固定的 CSS 重复背景?
我是开发 kafka-streams 应用程序的新手。我的流处理器旨在根据输入的 json 消息中的用户键值对 json 消息进行排序。
Message 1: {"UserID": "1", "Score":"123", "meta":"qwert"}
Message 2: {"UserID": "5", "Score":"780", "meta":"mnbvs"}
Message 3: {"UserID": "2", "Score":"0", "meta":"fghjk"}
我在这里读过Dynamically connecting a Kafka input stream to multiple output streams没有动态解决方案。
在我的用例中,我知道对输入流进行排序所需的用户键和输出主题。因此,我正在编写特定于每个用户的单独处理器应用程序,其中每个处理器应用程序都匹配不同的用户 ID。
所有不同的流处理器应用程序都从 kafka 中的同一个 json 输入主题读取,但如果满足预设的用户条件,每个应用程序只会将消息写入特定用户的输出主题。
public class SwitchStream extends AbstractProcessor<String, String> {
@Override
public void process(String key, String value) {
HashMap<String, String> message = new HashMap<>();
ObjectMapper mapper = new ObjectMapper();
try {
message = mapper.readValue(value, HashMap.class);
} catch (IOException e){}
// User condition UserID = 1
if(message.get("UserID").equals("1")) {
context().forward(key, value);
context().commit();
}
}
public static void main(String[] args) throws Exception {
Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "sort-stream-processor");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
TopologyBuilder builder = new TopologyBuilder();
builder.addSource("Source", "INPUT_TOPIC");
builder.addProcessor("Process", SwitchStream::new, "Source");
builder.addSink("Sink", "OUTPUT_TOPIC", "Process");
KafkaStreams streams = new KafkaStreams(builder, props);
streams.start();
}
}
问题一:如果使用低级处理器 API,是否可以使用高级流 DSL 轻松实现相同的功能? (我承认我发现很难理解和遵循高级流 DSL 的其他在线示例)
问题二:输入 json 主题正在以 20K-25K EPS 的高速率获取输入。我的处理器应用程序似乎跟不上这个输入流。我已经尝试部署每个进程的多个实例,但结果与我想要的结果相去甚远。理想情况下,每个处理器实例应该能够处理 3-5K EPS。
有没有办法改进我的处理器逻辑或使用高级流 DSL 编写相同的处理器逻辑?这会有什么不同吗?
最佳答案
您可以通过 filter()
在高级 DSL 中执行此操作(您有效地实现了一个过滤器,因为您只返回一条消息,如果它是 userID==1
)。您可以使用 KStream#branch()
来概括此过滤器模式(有关详细信息,请参阅文档:http://docs.confluent.io/current/streams/developer-guide.html#stateless-transformations)。另请阅读 JavaDocs:http://kafka.apache.org/0102/javadoc/index.html?org/apache/kafka/streams
KStreamBuilder builder = new KStreamBuilder();
builder.stream("INPUT_TOPIC")
.filter(new Predicate() {
@Overwrite
boolean test(String key, String value) {
// put you processor logic here
return message.get("UserID").equals("1")
}
})
.to("OUTPUT_TOPIC");
关于性能。单个实例应该能够处理 10K+ 条记录。如果没有任何进一步的信息,很难判断可能是什么问题。我建议在 Kafka 用户列表中询问(参见 http://kafka.apache.org/contact )
关于java - 使用 kafka-streams 有条件地对 json 输入流进行排序,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/43830457/
如何在 2014-10-04 - 2014-10-30 仅工作日和 08.00 - 20.00 之间随机更新日期列? 2014-10-04 - 2014-10-30 random working-da
我有一个二维 (3x7) 数组,我想转换为一维数组,以便我提供的行位于中心。行值可能沿途变化,但必须位于中心。 #define numRows 3 #define numCols 7 #define
我有2张 table : 第一个是“人”: person_id, 人名 第二个是“PersonsGraphs”: person_id1, person_id2, 关系类型 我正在寻找一种建立“家谱”的
是否可以在序列化 JSON 响应的同时根据 If 条件排除某些元素? if(a == 1) { //show element } else { //don't show element }
是否可以在序列化 JSON 响应的同时根据 If 条件排除某些元素? if(a == 1) { //show element } else { //don't show element }
尝试使用 jQuery 编写一个条件,该条件基本上说明,如果 div.gathering 不包含 a.cat-link,则执行以下操作。我已经尝试过以下方法,但似乎不起作用。有人能解释一下吗? if(
该练习要求插入值 x 的副本(这也是要在列表中搜索的值),但前提是该位置是另一个值 n 的倍数。未指定副本应插入到 x 值之前还是之后。 我的问题是并非在所有情况下都插入副本。我认为问题在于,当我插入
我遇到了这个[问题]:How can I store values into a multi-parameter struct and pass typedef struct to a functio
出于某种原因,当我编写 getWinner() 时,它仅适用于 2 种情况(最后一行)。就对角线和列而言,我拥有其他一切,但第 2 行(嗯,三,但数组,所以 2)基本上只适用于 o。只有当 o 位于
我有一个问题。 我想将“guid”列中的值复制到“帖子内容” 所有行都在一个表“wp-posts”中 “postparent”列中的一行有一个值,而“ID”列中的另一项也有相同的值 我必须做的事情是
我想将两个像这样的表合并到一个表中,并为重复的键行添加合并表中最旧的 DateAdded 值。 (Key1,Key2) 是主键。 +-----------+-----------+------+---
通过下面的表格和数据,我试图获得最高的 effective_from每个唯一 brand 小于当前时间戳的值/model组合 - 实际上是每件商品的当前价格。 CREATE TABLE things
您能告诉我如何删除未知号码的最后一条记录(有条件)吗? 例如,在这种情况下我想删除id为6到10的记录。 注意:该表和记录不是恒定的。 +----+-----+---------+ | id | ur
这个问题不太可能对任何 future 的访客有帮助;它只与一个较小的地理区域、一个特定的时间点或一个非常狭窄的情况相关,通常不适用于全世界的互联网受众。如需帮助使此问题更广泛适用,visit the
我有两个表, 标签 -> id,name,description,user,status 标签_连接。 -> id, Label_id, 类别 所以有多个类别,假设 1 => 新的,2 => 旧的。
好的,我会长话短说。 这是我的代码 String s = edittextkata.getText().toString(); String[] vowels = {"a","
我有一个非常具体的要求,我发现很难做到,我需要查找并替换文件中的某些行,但问题是文本不同,唯一的好处是它们都有一个 .[扩展名] 例如: 30/07/2012 14:46 17
我有一个大型数据库,其中存在各种不一致之处。我想澄清的项目之一是根据人口更改国家/地区状态。 数据样本是: { "_id" : "D", "name" : "Deutschland", "pop" :
我需要将范围(有条件)中的唯一值组合到同一行的另一个范围中。 其实我前两天发过类似的问题Link所提供的答案在我提出上述问题时有效。 但后来,我遇到了一个新问题,我宁愿问一个新的问题,让它更清楚: (
我刚开始使用 VBA,并且正在努力处理需要清理的工作表。 我有一列包含混合邮政编码和城市名称的字符串。我想从 A 列中提取邮政编码并放在 B 列中,并在 C 列中提取带有下划线的城市名称。 我的(示例
我是一名优秀的程序员,十分优秀!