问题一：Flink这种任务失败了，为什么还有1cu的任务在跑？
Flink这种任务失败了，为什么还有1cu的任务在跑？ 
job重启后，还多额外占用1cu的任务 
参考回答：
Apache Flink 是一个分布式流处理和批处理系统，它通过检查点机制来实现容错性。当Flink任务失败时，这通常意味着在执行过程中遇到了某种问题，比如节点故障、网络问题或者程序错误等。
如果你看到有1CU的任务仍在运行，可能的原因包括：
- 部分恢复：Flink的容错机制使得它可以在出现故障时尝试恢复任务。如果只是部分任务失败，并且系统的其他部分仍然能够正常工作，那么整个任务可能不会完全停止，而是继续运行剩余的部分。这种情况下，你可能会看到仍有1CU的任务在运行。
- 资源分配：在一些集群管理器（如YARN或Kubernetes）中，一旦任务被提交并且分配了资源，即使任务本身已经失败，这些资源可能仍会被占用。这意味着虽然任务已经失败，但它们仍然显示为正在运行，因为从集群的角度来看，那些资源已经被分配出去了。
- 未正确关闭的任务：有时，由于各种原因（例如编程错误或异常），任务可能没有正确地关闭。在这种情况下，尽管任务已经不再执行任何有用的工作，但它可能仍然显示为活动状态。
- 监控延迟：你的监控工具可能存在一定的延迟，也就是说，实际的任务可能已经停止，但是监视界面还没有更新这个信息。
- JobManager内存调整导致的问题：之前的信息提到过，增大JobManager的内存可能导致任务失败和YARN宕机。这也可能是影响任务状态的一个因素。
关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/573862
问题二：在Flink像这种既要split 又要where的，where条件放在那里？
在Flink像这种既要split 又要where的，where条件放在那里？ 
参考回答：
在Flink中，如果您需要同时执行split和where操作，可以按照以下方式进行处理：
- 使用
split()函数将流拆分为多个流。该函数接受一个OutputSelector参数，根据指定的条件将记录发送到不同的流中。例如：
DataStream<Integer> input = ...; SplitStream<Integer> splitStream = input.split(new OutputSelector<Integer>() { @Override public Iterable<String> select(Integer value) { List<String> outputNames = new ArrayList<>(); if (value % 2 == 0) { outputNames.add("even"); } else { outputNames.add("odd"); } return outputNames; } });
- 在上面的示例中，根据输入整数值的奇偶性，我们将其拆分为"even"和"odd"两个流。
- 对拆分后的流应用
filter()函数来实现where条件过滤。这将对每个流应用过滤器，只保留满足指定条件的记录。例如：
DataStream<Integer> evenStream = splitStream.select("even"); DataStream<Integer> filteredStream = evenStream.filter(value -> value > 10);
关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/573860
问题三：flink 1.15 是否可以在 jdk1.8下运行么？还是必须用 jdk 11
flink 1.15 是否可以在 jdk1.8下运行么？还是必须用 jdk 11
参考回答：
可以，云上都还是。
关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/573859
问题四：在Flink用CEP的话需要怎么整理的呢？
在Flink用CEP的话需要怎么整理的呢？假设我有一组数据a1 a2 a3 a4 a5 a6 b
我想筛出来下面的这个样子
a1 a2
a1 a3
a1 a4
a1 a5
a1 a6
a1 b
或者
a1 a2
a1 a2 a3
a1 a2 a3 a4
a1 a2 a3 a4 a5
a1 a2 a3 a4 a5 a6
a1 a2 a3 a4 a5 a6 b
参考回答：
在Flink中使用CEP（Complex Event Processing）库来实现您的需求，您可以按照以下步骤进行整理：
- 首先，您需要将输入数据转换为Flink的
DataStream对象。假设您的数据是一个字符串列表，可以通过fromCollection()方法将其转换为DataStream。例如：
List<String> input = Arrays.asList("a1", "a2", "a3", "a4", "a5", "a6", "b"); DataStream<String> dataStream = env.fromCollection(input);
- 接下来，您需要定义CEP模式并应用于数据流上。根据您的需求，有两种可能的模式。
a) 要筛选出"a1"与后续记录的组合，您可以使用严格近邻模式（Strict Contiguity）。此模式要求事件按顺序连续出现。
Pattern<String, ?> pattern = Pattern.<String>begin("a1").followedByAny().where(new SimpleCondition<String>() { @Override public boolean filter(String value) { return !value.equals("b"); } });
- b) 要筛选出"a1"与后续记录的逐渐增长的组合，您可以使用宽松近邻模式（Relaxed Contiguity）。此模式允许事件之间存在间隔。
Pattern<String, ?> pattern = Pattern.<String>begin("a1").followedByAny( Pattern.<String>begin("a1").followedByAny().where(new SimpleCondition<String>() { @Override public boolean filter(String value) { return !value.equals("b"); } }) );
- 应用CEP模式并提取匹配的结果。在Flink中，您可以使用
CEP.pattern()方法将模式应用于数据流，并使用select()方法选择特定的结果。例如：
PatternStream<String> patternStream = CEP.pattern(dataStream, pattern); DataStream<String> resultStream = patternStream.select((Map<String, List<String>> patternMatch) -> { StringBuilder sb = new StringBuilder(); sb.append("a1"); for (String key : patternMatch.keySet()) { sb.append(" ").append(key); } return sb.toString(); });
关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/573858
问题五：相同的程序一个在云flink产品部署时Managed Memory打满，另一个几乎没有占用，为什么？
相同的两个程序一个在云flink产品上部署时 Managed Memory 直接打满，另一个在emr自建的yarn上跑 Managed Memory 几乎 没有占用，这是怎么回事？ 
参考回答：
managed memory 一启动Gemini就会把所有分配给它的managed都claim过去，所以这上面是看不出实际用量的。
关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/573857
