什么是java/scalakafka流相当于KSQL加入WHERE子句？

Question

假设我有两个Kafka流（Kafka流scala库，版本2.2.0）：

val builder: StreamsBuilder = new StreamsBuilder
val stream1: KStream[String, GenericRecord] = builder.stream[String, GenericRecord]("topic1")
val stream2: KStream[String, GenericRecord] = builder.stream[String, GenericRecord]("topic2")

以及他们的加入：

val stream3: KStream[String, MyClass] = flights.join(schedules)((r1, r2) =>  MyClass(r1.get("f1"), r2.get("f2")), JoinWindows.of(Duration.ofSeconds(30))

KSQL中WHERE子句的等价物是什么？（参见最新订单流）了解流API？使用stream3是个好主意。滤器这种方法的效率是否与KSQL创建的流相同？

Answer 1

对于流API，什么是KSQL中可用的WHERE子句？（参见late_orders流）？

它是：

• KStream#filter（），它返回一个经过过滤的KStream
• KTable#filter（），返回一个经过过滤的KTable

https://kafka.apache.org/documentation/streams/developer-guide/dsl-api.html#stateless-转变

使用stream3是个好主意。滤器

是的。

这种方法的效率是否与KSQL创建的流相同？

是的。