问题标签 [flink-cep]

如果使用 Flink CEP 库找到 Hello 和 world，我正在尝试打印出一个字符串。我的来源是 Kafka，并使用控制台生产者输入数据。那部分正在工作。我可以打印出我进入主题的内容。但是，它不会打印出我的最后一条消息“世界真好！”。它甚至不会打印出它进入了 lambda。下面是类

任何帮助将不胜感激。

谢谢！

我正在研究用于数据流的 apache flink，我有几个问题。任何帮助是极大的赞赏。谢谢。

1）创建翻滚窗口是否有任何限制。例如，如果我想为每个用户 ID 创建一个持续 2 秒的滚动窗口，假设我有超过 1000 万个用户 ID，那将是一个问题。（我正在使用 keyBy 用户 ID，然后创建一个 timeWindow 2 秒）？这些窗口在 flink 内部是如何维护的？

2）我查看了循环分区的重新平衡。假设我设置了一个集群，如果我的源并行度为 1，并且如果我进行重新平衡，我的数据是否会在机器之间进行混洗以提高性能？如果是这样，是否有特定的端口用于将数据传输到集群中的其他节点？

3) 状态维护有什么限制吗？我计划维护一些可能会变得非常大的用户 ID 相关数据。我读到了 flink 使用 Rocks db 来维护状态。只是想检查是否对可以维护多少数据有任何限制？

4）如果数据量较少，状态在哪里维护？（我猜在 JVM 内存中）如果我的集群上有几台机器，每个节点都可以获得当前状态版本吗？

我的要求是在一天内流式传输数百万条记录，并且它对外部配置参数有很大的依赖性。例如，用户可以随时在 Web 应用程序中更改所需的设置，并且在进行更改后，必须使用新的应用程序配置参数进行流式传输。这些是应用程序级别的配置，我们还有一些动态排除参数，每个数据都必须通过和过滤。

我看到 flink 没有在所有任务管理器和子任务之间共享的全局状态。拥有一个集中式缓存是一种选择，但对于每个参数，我都必须从缓存中读取它，这会增加延迟。请就处理此类场景的更好方法以及其他应用程序如何处理它提出建议。谢谢。

我正在尝试使用 Apache Flink 构建一个 CEP 系统来关联事件。要求之一是能够在运行时为异常检测添加新模式，而不会丢失系统可用性。关于我该怎么做的任何想法？

例如，如果我有一系列安全事件（例如访问、身份验证）和用于检测异常的模式（例如，1 分钟内登录同一台机器超过 10 次）我希望能够更改模式参数，例如而不是 10 次登录，也许我想要 8 次，同时我希望能够创建其他模式（维护相同的流）来检测新类型的异常，而不会丢失事件/系统可用性。

问候。

我们计划使用 Flink CEP 根据一些动态模板来处理大量的事件。系统必须识别事件链（有时是具有条件和分组的复杂链）。模板将由用户创建。换句话说，我们必须在不接触代码的情况下创建复杂的模板。是否可以使用 Apache Flink 来解决这个问题？Filnk 是否支持动态模板？

CEP 库使得编写一个程序来描述它可以很容易地判断 X 是否跟在 Y 后面。

val pattern = Pattern.begin("start").where(_.value == "X") .next("end).where(_.value == "Y")

但是我将如何描述一个我不关心 X 和 Y 的顺序的程序，只关心一个跟随另一个的顺序。例如 X, Y 和 Y, X 都很有趣。但是，X,X 或 Y,Y 并不有趣，所以我无法匹配第三个共享属性。

我的代码如下：

...定义我的模式

这段代码可以工作并且可以做我想做的事，我得到一个遵循我设置的模式的结果流。

我想知道的是是否可以将新模式应用于我稍后添加到环境中的源，从而获得与不同模式匹配的不同结果流，而无需再次调用 env.execute() 因为当我这样做时除了我的新结果流之外，我还得到了多余的旧结果流（即旧模式被多次执行）？

我有两个事件流：一个发出事件以表示项目生命周期的开始，另一个流发出事件以表示项目生命周期的结束。（流可以在 . 上加入itemId。）

如何在 Flink 中为每个只有itemId1“生命周期结束”事件而不是相应开始的事件发出一个新事件？（这些开始和结束的事件可能相隔数小时或数天。）

如果两个事件流入 Flink，它们是否可以使用随后的第三个事件中的信息进行逻辑连接（使用 DataStream API 或 CEP）？例如，下例中的第三个事件能否用于根据其 right_id 和 left_id 链接前两个事件？

如果模式定义只有属性级别的比较， Flink-CEP 是否使用hashCode()andequals()进行比较？

我问是因为我不能保证hashCode()正确性。