问题标签 [flink-batch]

让我解释一个我需要处理的场景。让我们假设三个设备 A、B、C 正在向 flink CEP 发送日志进行处理。让我们假设模式为 A，然后是 5 分钟 B，然后是 5 分钟后的 C。让我们假设 B 设备停机并在 50 分钟后发送日志的场景。所以在这种情况下，所有事件都将被丢弃。我只是想知道 flink 中是否有任何支持将状态维持在特定定义的时间间隔（假设在我的情况下为 1 天，这意味着 A 和 C 日志将存储 1 天后日志将是在不匹配的情况下丢弃）。请从 CEP 的角度提出可行性。

是否可以在 flink 中创建无界的集合流。就像在地图中一样，如果我们添加一个元素，flink 应该像在套接字流中一样处理。一旦读取了初始元素，它就不应退出。

我有一个用例，其中大量日志将被 apache flink CEP 消耗。我的用例是找到蛮力攻击和端口扫描攻击。这里的挑战是，在普通 CEP 中，我们将值与“event” = login 之类的常量进行比较。在这种情况下，标准是不同的，因为在蛮力攻击的情况下，我们有如下标准。

用户名是常量并且 event="login failure" （分隔事件在 5 分钟内发生 5 次）。这意味着在 5 分钟内收到 5 次相同用户名的登录失败事件的日志

对于端口扫描，我们有以下标准。

ip 地址是常量，dest 端口是可变的（分隔符是事件在 1 分钟内发生 10 次）。这意味着在 1 分钟内收到了 10 个不同端口的具有恒定 IP 地址的日志。

我试图了解 Flinks Collector.collect() 的作用以及它如何处理传入/传出数据：

取自 Flink DataSet API 的示例：

以下代码将文本行的 DataSet 转换为单词的 DataSet：

因此，文本 Lines 被拆分为标记，每个标记都被“收集”。听起来很直观，但我错过了 Collector.collect() 背后的实际动态。收集到的数据在分配给输出之前存储在哪里，即 Flink 是否将它们放入某种 Buffer 中？如果是，数据如何传输到网络？

我仍在苦苦思考 flink 如何在不同运营商之间“交换/传输”数据，以及运营商之间的实际数据会发生什么。

以上面的 DAG 为例： 执行的 DAG

1. DataSet 被转发/传输到 GroupReduce Operator 的所有并行实例，Data 根据 GroupReduce 转换得到缩减。

2. 所有新数据都被转发到 Filter->Map->Map Operand，即 GroupReduce 运算符的并行实例之一消耗的所有数据都被转移到 Filter->Map->Map 运算符的一个实例（不需要用于序列化/反序列化，因此 Operator 访问由 GroupReduce Operator 生成的数据）

3. 所有GroupReduces 输出数据都经过哈希处理，并在 (Filter->Map) 运算符的所有并行实例之间均匀分布/传输（运算符之间需要序列化/反序列化）

因此，例如，如果 GroupReduce Operators 输出大约为 100MB，它会将 100MB 转发到 (Filter->Map->Map) 操作数，然后散列该 100MB 的副本并将其传输到 (Filter->Map) 实例。所以我会再产生 100MB 的网络流量

我很困惑为什么在 GroupReduce 之后和过滤步骤之前有这么多网络流量。在将现在减少的数据发送给后续操作员之前将 GroupRedcue 和 Filter 步骤链接在一起会更好吗？

我对 Flink 的行为有疑问。下面是我的代码片段。如您所见，某些服务正在提供 Flink 将要执行的 sql 条件列表（例如大约 10k sql）。我的问题是，每当更新 sql 时，我如何指示 flink 使用新的 sql？我看到的一种方法是停止和启动我想避免的 flink 服务，因为其他 sql 条件需要一直运行，并且只有正在更新的条件需要动态停止/启动/或更新。另外，我不想将 10k sqls 作为 10k 个不同的作业提交。那么我正在寻找的行为是否可以使用 Flink 1.11 版？

Spark RDD.persist(..) 可以帮助避免重复的 RDD 评估。

Flink 中是否有相同的功能？

其实我想知道如果我像下面这样编码，Flink 会评估dataStream一次或两次？

目前，我有一个流式作业，它在收到特定触发器时触发批处理作业。

我想关注那个被触发的批处理作业，当它完成时，想在数据库中插入一个条目，比如弹性搜索等。

任何想法，我们如何实现这一目标？我们怎么能听那份工作呢？

我正在尝试在集群上运行 flink 应用程序。应用程序部署成功，我可以看到 jobmanger 和 taskmanager 正在运行并且资源注册成功。

应用程序需要虚拟事件，它工作正常，sql 查询也返回结果（通过任务管理器）。

问题：应用程序从消息集线器接收事件，但这些消息仅与作业管理器一起使用，并到达任务管理器以在集群（独立）中执行。但它适用于 intelliJ。

我知道它是否会有 kafka 然后 addSource(new FlinkKafkaConsumer()) 工作，但对我来说，我必须注册回调，然后消息出现 onEvent。

是否有任何机制可以帮助从集群中的作业管理器向任务管理器发送事件？

我想使用多个 KeySelector 函数基于多个值对 POJO 数据集进行排序：

这会产生“KeySelector cannot be chained”错误。根据 Flink 文档，应该可以链接 sortPartition 函数。

有没有办法在不使用字段表达式的情况下解决这个问题？