apache-flink - Flink 中的处理流
问题描述
我需要按如下方式处理消息:
- 每条消息必须不变地通过所有进程。
- 如果进程检测到模式匹配,则必须将消息的副本传递给另一个线程。
streamIn -> process1 -> process2 -> .... -> processN
\ | /
\ | /
sink to kafka
我尝试使用 OutputTag
OutputTag<SysmonPartial> filtersOutput = new OutputTag<SysmonPartial>("FiltersOutput"){};
DataStream<SysmonPartial> kafkaSource = env.addSource(consumer);
DataStream<SysmonPartial> source = kafkaSource.rebalance();
SingleOutputStreamOperator<SysmonPartial> s = source
.process(lambda1).name("lambda1").startNewChain()
.process(lambda2).name("lambda2").startNewChain()
.process(lambda3).startNewChain()
.process(lambda4).startNewChain()
.process(lambda5).startNewChain()
.process(lambda6).startNewChain()
.process(lambda7).startNewChain();
SingleOutputStreamOperator<Any> output = s
.getSideOutput(filtersOutput)
.process(filterProcessFunction).setParallelism(1).startNewChain();
output.addSink(sink).setParallelism(1);
env.execute(jobName);
其中 lambda 1、2 等是条件检查函数,例如
public class Bypass_WS_01_03 extends FilterTagFuction<SysmonPartial, SysmonPartial, SysmonPartial> {
private static final Pattern p_1 = Pattern.compile("pattern1");
private static final Pattern p_0 = Pattern.compile("pattern2");
@Override
public void processElement(SysmonPartial t, Context ctx, Collector<SysmonPartial> out) throws Exception {
out.collect(t);
if (
"1".equals(t.B_VendorEventID) &&
t.CommandLine != null && t.CommandLine.length() != 0 && p_0.matcher(t.CommandLine).find() &&
t.ImageName != null && t.ImageName.length() != 0 && p_1.matcher(t.ImageName).find()
) {
t.RuleId = "Bypass_WS_01_03";
ctx.output(getOutputTag(), t);
}
}
}
但是由于某种原因它对我不起作用,也许还有其他方法?正如我从文档中了解到的(https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/side_output.html),OutputTag用于创建标有 Tag 的附加消息。还是我错了?
解决方案
您连接作业图的方式意味着在您访问侧面输出的作业结束时
SingleOutputStreamOperator<Any> output = s
.getSideOutput(filtersOutput)
你只会得到最后一个进程函数放在侧面输出上的任何东西——即只有 lambda7 发出的事件。
我相信你打算做的可以表示为
SingleOutputStreamOperator<...> s1 = source.process(lambda1).name("lambda1");
SingleOutputStreamOperator<...> s2 = s1.process(lambda2).name("lambda2");
SingleOutputStreamOperator<...> s3 = s2.process(lambda3);
...
DataStream<...> side1 = s1.getSideOutput(filtersOutput);
DataStream<...> side2 = s2.getSideOutput(filtersOutput);
DataStream<...> side3 = s3.getSideOutput(filtersOutput);
...
SingleOutputStreamOperator<Any> output = side1.union(side2, side3, ...)
.process(filterProcessFunction)
...
另外,我认为您不应该使用startNewChain()
(这会禁用运算符链接)。运算符链接是一项有价值的优化,仅应在特殊情况下禁用。
推荐阅读
- sql - 如何从初始大表创建星型模式?
- plotly - 如何用plotly js制作一个垂直于x轴的平面
- c# - C# CRUD 问题 - UPDATE 面板无法按预期工作
- angular - 如何在 Angular 中同时发布多个 POST 请求?
- json - Spring Boot - 在 URI 处读取 CSV 文件并将其作为 JSON 返回
- python - 发送到多处理时我的对象是否被复制
- c++ - 迭代未遍历集的向量
- java - 使用 testNG 框架在数据提供者中获取异常
- python - 如何在 Python 中读取电影的 exif 数据?
- forms - 如何在注册表单中强制设置角色单选按钮。德鲁巴 8