如何使用Kafka消费者为Flink CEP编写Junit测试代码

文章/答案/技术大牛

发布

1回答

apache-kafka、apache-flink、flink-cep

我们有一个execute()方法，其中我们使用FlinkKafkaConsumer08作为我们的Flink CEP源，然后我们有CEP模式，并且alert再次进入另一个kafka主题。如何为这个execute()方法编写一个junit测试用例？有没有人能为我提供一个样例junit代码？ Pattern.

浏览 37提问于2019-01-02得票数 0

回答已采纳

0回答

如何从程序中停止flink流作业

junit、apache-kafka、apache-flink、flink-streaming

我正在尝试为Flink streaming作业创建一个JUnit测试，该作业分别使用FlinkKafkaProducer09和FlinkKafkaConsumer09将数据写入kafka主题并从相同的kafka我在produce中传递了一个测试数据： DataStream<String> stream = env.fromElements("tom", "jerry", "bill"

浏览 5提问于2017-06-09得票数 9

2回答

IntelliJ集成开发环境警告Flink CEP未知错误

scala、intellij-idea、apache-flink、complex-event-processing

我开始研究用Scala语言编写的Apache Flink的CEP库，当我试图通过执行https://ci.apache.org/projects/flink/flink-docs-stable/dev这是我写的代码。我知道它不完整，但自从这个问题出现后，我无论如何都不能完成它。/artifact/org.apache.flink/flink-co

浏览 50提问于2020-01-14得票数 0

回答已采纳

1回答

Flink，在使用多个Kafka源时，如何正确设置并行性？

apache-flink、flink-streaming

我仍然不能清楚地理解并行性，比方说，我们有一个flink集群，它有足够的插槽。在我们的flink工作中，我们从3个不同的kafka集群中使用3个kafka主题，每个主题有10个分区。如果我们想尽快使用该消息。并行性应该是什么？如果我们将并行性设置为10，那么就会使用10个槽，这意味着，根据我的理解，10个线程对吗？如果我们将并行性设置为30，那么就有30个线程，但这30个线程是否足够聪明，它们中的10个会转到topic_1，10个会

浏览 0提问于2019-03-13得票数 12

回答已采纳

1回答

如何在flink sql表中自动生成水印？

apache-flink、flink-sql、flink-cep

我正在测试flink cep，我的水印被定义为行时间，我的表是一个kafka表。由于水印依赖于所有kafka分区的最小值，所以每一条新消息都必须等待kafka分区对齐，然后cep触发结果。我的kafka表(主题有3个分区)被定义为 agent_id String, room_id String, create_timeto_timestamp_ltz

浏览 4提问于2021-10-30得票数 0

回答已采纳

1回答

如何在Flink表API中加入两个数据流和滑动窗口函数？

apache-flink、flink-streaming、apache-zeppelin、flink-sql、amazon-kinesis-analytics

我有两个来自两个Kafka主题的流表，我想加入这些流并对所连接的数据执行聚合功能。需要使用滑动窗口连接流。下面是代码片段 ep.event_flink_time, ep.rowtime,TIMESTAMPDIFF(SECOND, ep.event_flink_time, cep.event_flink</em

浏览 25提问于2022-05-24得票数 0

2回答

如何在spring boot测试中禁用kafka连接？

spring-boot、spring-test、spring-kafka

我使用spring kafka消费来自kafka主题的消息，所以我有一个kafka消费者配置类： @Configuration // kafkaconsumer configurations } 我有一些JUnit测试，会用mockMvc加载spring上下文来测试我的API，我不想测试与kafka消息相关的功能，我

浏览 115提问于2019-08-29得票数 1

1回答

Apache Flink* CEP，模式不匹配*

java、apache-flink、flink-cep

我是新的Flink CEP，并试图测试基本的东西-在下面的代码，我的期望是所有的输入应该匹配模式，并应打印为匹配的结果。package com.o9.flink;import org.apache.flink.cep.CEP;import org.apache.flink.cep

浏览 9提问于2022-08-09得票数 0

回答已采纳

1回答

生成“假”流数据。卡夫卡

java、apache-kafka、apache-flink、stream-processing、event-stream-processing

我试图生成流数据，以模拟在不同的时间范围内接收两个值( Integer类型)的情况，使用时间戳和Kafka作为连接器。如果是，请让

浏览 2提问于2018-08-20得票数 2

2回答

如何使用bazel使用/导入Flink的TestHarness类？

maven、testing、apache-flink、bazel、flink-streaming

我的直觉可能是正确的使用包装和/或分类器。为了为构建文件找到正确的导入标记，我建议查看.cache/bazel/bazel_user/.../external/maven/BUILD file我想使用Bazel构建工具测试我自己的异步flink流操作符RichAsyncFunction。这与的问题基本相同主要的问题是，我找不到导入的org.apache.flink.streaming.ut

浏览 0提问于2020-05-09得票数 1

1回答

测试kafka和flink集成流程

scala、apache-kafka、integration-testing、apache-flink

例如，我想测试卡夫卡/ Flink与FlinkKafkaConsumer011和FlinkKafkaProducer011的集成。这一进程将是：当我说测试时，这是单元/集成测试。谢谢!

浏览 0提问于2018-04-16得票数 3

1回答

MapFunction的实现不是可序列化的Flink

java、serialization、apache-flink、flink-streaming

我正在尝试实现一个类，使用户能够操纵N个输入流，而不对输入流的类型有限制。我总是遇到序列化的问题，我试着遵循这个指南，但它没有工作。在Java中什么是序列化/反序列化？SerializationWhat在Flink中可以解决哪些问题是我的代码中的问题(您可以在代码和错误消息下面找到) 非常感谢。dataStrea

浏览 19提问于2020-04-09得票数 2

回答已采纳

1回答

从Kafka主题反序列化FlinkKafkaProducer的事件会导致空的JSON记录

json、serialization、apache-kafka、deserialization、apache-flink

编辑i编写了最小可重现性示例我使用kafka-console-consumer.sh检查从Flink作业发送到Kafka主题的内容，但我只看到空的JSON记录，而我期待的是非

浏览 5提问于2021-08-11得票数 0

回答已采纳

1回答

流在Hadoop中的应用

hadoop、spark-streaming、apache-flink、flink-streaming、flink-cep

现在，我知道，如果我使用Kafka，那么我需要两个应用程序；一个用于将数据摄取到Kafka主题中，另一个用于消费Kafka主题的数据。我有几个问题我没有得到回答：如果像卡夫卡/动态)这样的流层出现在两者

浏览 2提问于2016-06-13得票数 0

2回答

Apache Flink: Python流API中的Kafka连接器，“无法加载用户类”

python、apache-kafka、apache-flink、flink-streaming、pyflink

/flink-1.6.1/bin/pyflink-stream.sh examples/read_from_kafka.py运行我的脚本。python脚本相当简单，我只是尝试使用现有主题中的内容，并将所有内容发送到标准输出(或日志目录中的*.out文件，默认情况下，输出方法将在该文件中发出数据)。-0.9_2.11-1.6.1.jar、flink-connector-kafka-base_2.11-1.6.1.jar和kafka-clients-0

浏览 0提问于2018-10-10得票数 4

1回答

手动重置kafka偏移量时，Flink偏移量进入不一致状态

apache-kafka、flink-streaming

我们有一个flink流应用程序，从卡夫卡读取消息。由于某些原因，我们不得不从kafka重置命令将kafka偏移重置为最新，因为有大量堆积。我们希望flink应用程序跳过所有这些消息，并从重置后出现的新消息开始。问题是因为flink在内部管理它的偏移量，它不知道这个重置，它现在只从后向读取msg (重置前的偏移点)，现在也不能提交偏移量。因此，每次重启flink应用程序时，它都会再次从同一点读取。我知道我们不应该在flink kafka

浏览 62提问于2021-06-19得票数 4

回答已采纳

2回答

Flink与合流Kafka模式注册表

apache-kafka、apache-flink、avro、confluent-schema-registry

我正试图使用FlinkKafkaProducer10从Flink中将kafka与模式注册表汇合在一起。产生下面的错误。代码编译但会产生运行时错误。生成错误的完整代码示例在上。所有资源都是模拟的，kafka集群和模式注册表。代码实际上是测试用例。) at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducerBase.<init>(Fli

浏览 4提问于2017-11-19得票数 4

1回答

带有重试模板的Apache侦听器的Junit

java、spring-boot、junit、apache-kafka、spring-kafka

需要指导如何为Kafka消费者类编写集成JUnit测试，以断言消息消费被重试了多少次。我在Kafka配置中添加了一个重试模板。retryTemplate.setRetryPolicy(retryPolicy);} @KafkaListener(topics = "${kafka.topic}", groupId = "${kafka.grou

浏览 2提问于2021-07-15得票数 0

1回答

使用Apache创建CEP

apache-kafka、apache-flink、flink-streaming、flink-cep

我试图为Kafka InputStream实现一个非常简单的Apache。卡夫卡制作者创造了一个简单的双重价值，并通过卡夫卡主题发送给消费者。目前，我正在用Flink编写CEP消费者的代码。到目前为止，这是我的书面代码： StreamExecutionEnvironmentproperties.setProperty(&

浏览 1提问于2018-09-13得票数 0

回答已采纳

点击加载更多