问题标签 [flink-cep]

在 Esper CEP 引擎事件模式中，有一个Every操作符指示模式子表达式应该在每个关键字限定的子表达式计算为真或假时重新启动。

根据此文档，我们可以通过every不同的方式使用运算符，例如：

我想知道在 Flink CEP 中哪些模式是可能的？我想every A -> B在 Flink 中使用，但不知道如何。

而 Flink 中这个默认的 CEP 模式代表了上述哪种模式？（使用 Flink v1.1.4）

提前致谢。

subtype(Tuple.getTupleClass( 2 ))，并出现错误 Inferred type 'capture<? extends org.apapche.flink.api.java.tuple.Tuple>' for type parameter 'S' is not within its bound;should extend 'org.apapche.flink.api.java.tuple.Tuple3<java.lang.String,java.lang.String,java.lang.String>'

我应该修改这个吗？但是如何修改？Pattern< Tuple3< String, String, String >, ? > pattern

由 2017012 更新

我试过这个，不应该填写什么。PatternStream< ...>感谢任何可以提供帮助的人。

我试图找出如何实现以下具有事件流的逻辑：

1. 第一个事件出现在给定的时间窗口（30 秒）
2. 与第一个事件相关的第二个事件（通过某些属性）没有出现在给定的时间窗口中
3. 生成新的错误事件

目前我实现了非常简单的“相反”实现：

如果能够根据事件的字段值匹配事件，这将超出当前从匹配单独条件的事件中创建模式的能力。例如，如https://flink.apache.org/news/2016/04/06/cep-monitoring.html中所述，我们可以这样做：

但是，使用以下功能创建 Pattern 会很棒：.where(second_evt->evt.getTemperature() == first_evt->evt.getTemperature()

I created a HA Flink v1.2 cluster made up of 1 JobManager and 2 TaskManagers each in its own VM (not using YARN or hdfs). After I start a job on the JobManager node I kill one TaskManager instance. Immediately in the Web Dashboard I can see the job being cancelled and then failing. If I check the logs:

In the job implementation I have

My question is shouldn't the JobManager automatically redirect all requests to the remaining / running TaskManager? Similarly if I start the JobManager and 1 TaskManager instance, and run a job, when I start the 2nd TaskManager instance should it also contribute to solve the running job?

Thanks!

我有一个 Flink v1.2 设置，有 1 个 JobManager，2 个 TaskManager，每个都在它自己的 VM 中。我将状态后端配置为文件系统，并在上述每个主机的情况下将其指向本地位置（state.backend.fs.checkpointdir：file:///home/ubuntu/Prototype/flink/flink-checkpoints）。我已将并行度设置为 1，每个任务管理器都有 1 个插槽。然后我在 JobManager 上运行一个事件处理作业，将其分配给 TaskManager。我杀死了运行作业的 TaskManager，在失败的 TaskManager 上尝试了几次失败后，Flink 尝试在剩余的 TaskManager 上运行作业。此时它再次失败，因为它找不到相应的检查点/状态：java.io.FileNotFoundException：

文件夹 /home/ubuntu/Prototype/flink/flink-checkpoints/56c409681baeaf205bc1ba6cbe9f8091 仅存在于我杀死的 TaskManager 而不是另一个。

我的问题是，如果我想要上述功能，是否应该在所有任务管理器上为检查点/状态设置相同的位置？

谢谢！

我已经在本地以 HA 集群模式 2 JobManagers 1 TaskManager 安装了 Flink 1.2，但它一直拒绝以显示“正在启动集群”的模式实际启动。消息而不是“在 ZooKeeper 仲裁中启动具有 2 个主节点和 1 个对等节点的 HA 集群”。

显然在 bin/config.sh 它读取配置如下：

这意味着独立于配置文件中为“高可用性”键配置的内容（在我的情况下，值是“zookeeper”）它将设置为“无”并在 bin/start-cluster.sh

永远不会进入第一个 if 分支。

还有其他人遇到过这个吗？

我试图了解 Flink 是否可以按照我需要的方式执行。

使用 CEP 库进行模式匹配，是否可以根据初始模式匹配设置多个流？

例如，一条消息进来并匹配模式 x1，然后下一条消息匹配模式 y1，接下来是 y2，下一个是 x2。Flink 有没有办法区分多组模式，并在指定的时间长度内不断寻找每个模式匹配？

还是 Flink 只是忽略与模式下一部分不匹配的消息？如果顺序是 x1 > y1 > y2 > x2，是否会因为不匹配 x1 的模式而忽略 y1 和 y2？

我有一个 Flink v1.2、3 个 JobManagers、2 个 TaskManagers 的设置。我想为后端状态和检查点以及 zookeeper storageDir 使用 S3 存储桶而不是 hdfs

fs.s3.accessKey：[accessKey]
fs.s3.secretKey：[secretKey]

state.backend：文件系统

state.backend.fs.checkpointdir: s3:///[bucket]/flink-checkpoints
state.checkpoints.dir: s3:///[bucket]/external-checkpoints
高可用性：zookeeper
high-availability.zookeeper.storageDir : s3:///[桶]/recovery

在我登录的 JobManager 中，我有

我没有安装hadoop。不确定这是否需要以及是否应该如何/在哪里安装/配置？

编辑：在使用以下 hadoop xml (core-site.xml) 配置 Flink 后，我并没有真正理解 IAM 部分并且我没有使用 EMR，我自己安装了集群（在 AWS 中）以便能够更新 Flink 而无需依赖图片：

我收到此错误：

编辑：我的错误是我在描述字段中设置了键而不是值。

我有一个 Flink v1.2、3 个 JobManagers、2 个 TaskManagers 的设置。我想将 hdfs 用于后端状态和检查点以及 zookeeper storageDir

state.backend：文件系统

state.backend.fs.checkpointdir: hdfs:///[ip:port]/flink-checkpoints
state.checkpoints.dir: hdfs:///[ip:port]/external-checkpoints
高可用性：zookeeper
高可用性.zookeeper.storageDir: hdfs:///[ip:port]/recovery

在我登录的 JobManager 中，我有

Hadoop 作为单节点集群安装在我在设置中设置的 VM 上。为什么 Flink 要求配置额外的参数？（顺便说一句，它们不在官方文档中）