从自定义源写入数据以连续方式瞳孔

Question

这是我第一次使用 Apache Flink (1.3.1( 并有一个问题。更详细地说，我正在使用 flink-core、flink-cep 和 flink-streaming 库。我的应用程序是一个 Akka Actor系统，它使用来自 RabbitMQ 的消息，并且各种参与者处理这些消息。在某些Actor中，我想实例化来自Flink的StreamExecutionEnvironment并处理传入的消息。因此，我编写了一个自定义源类来扩展RichSourceFunction类。一切都能找到，除了一件事：我不知道如何将数据发送到我的 Flink 扩展。这是我的设置：

public class FlinkExtension {
private static StreamExecutionEnvironment environment;
private DataStream<ValueEvent> input;
private CustomSourceFunction function;
public FlinkExtension(){
environment = StreamExecutionEnvironment.getExecutionEnvironment();
function = new CustomSourceFunction();
input = environment.addSource(function);
PatternStream<ValueEvent> patternStream = CEP.pattern(input, _pattern());
DataStream<String> warnings = patternStream.select(new PatternSelectFunction<ValueEvent, String>() {
@Override
public String select(Map<String, List<ValueEvent>> pattern) throws Exception {
return null; //TODO
}
});
warnings.print();
try {
environment.execute();
} catch(Exception e){
e.printStackTrace();
}
}
private Pattern<ValueEvent, ?> _pattern(){
return Pattern.<ValueEvent>begin("first").where(new SimpleCondition<ValueEvent>() {
@Override
public boolean filter(ValueEvent value) throws Exception {
return value.getValue() > 10;
}
});
}
public void sendData(ValueEvent value){
function.sendData(value);
}
}

这是我的自定义源函数：

public class CustomSourceFunction extends RichSourceFunction<ValueEvent> {
private volatile boolean run = false;
private SourceContext<ValueEvent> context;
@Override
public void open(Configuration parameters){
run = true;
}
@Override
public void run(SourceContext<ValueEvent> ctx) throws Exception {
this.context = ctx;
while (run){
}
}
public void sendData(ValueEvent value){
this.context.collectWithTimestamp(value, Calendar.getInstance().getTimeInMillis());
}
@Override
public void cancel() {
run = false;
}
}

所以我想从外部调用FlinkExtension类中sendData的方法，以连续的方式将数据写入我的FlinkExtension。这是我的 JUnit 测试什么应该将数据发送到扩展，然后将数据写入SourceContext。

@Test
public void testSendData(){
FlinkExtension extension = new FlinkExtension();
extension.sendData(new ValueEvent(30));
}

但是如果我运行测试，什么也没发生，应用程序挂起在CustomSourceFunction的运行方法中。我还尝试在CustomSourceFunctionrun 方法中创建一个新的无限线程。

总结一下：有没有人知道如何以连续的方式将数据从应用程序写入 Flink 实例？

Answer 1

Flink 源连接器通过让它们的 run(( 方法在 while(run( 循环中调用 collect(((或 collectWithTimestamp(((来发出连续的数据流。如果你想研究一个例子，Apache NiFi源并不像大多数那样复杂;这是它的运行方法。

Answer 2

问题是run方法和sendData方法使用不同的CustomSourceFunction对象实例。因此，context对象不在方法之间共享，并且添加新ValueEvent不起作用。

若要解决此问题，请将run方法使用的对象实例存储为CustomSourceFunction类的静态成员变量。 当您需要创建新ValueEvent时，在同一对象实例上调用sendData方法。

请参阅下面的示例代码

package RuleSources;
import Rules.Rule;
import org.apache.flink.streaming.api.watermark.Watermark;
import java.util.ArrayList;
public class DynamicRuleSource extends AlertingRuleSource {
private static DynamicRuleSource sourceObj;
private SourceContext<Rule> ctx;
public static DynamicRuleSource getSourceObject() {
return sourceObj;
}
public void run(SourceContext<Rule> ctx) throws Exception {
this.ctx = ctx;
sourceObj = this;
while(true) {
Thread.sleep(100);
}
}
public void addRule(Rule rule) {
ctx.collect(rule);
}
@Override
public void cancel() {
}
}

添加新规则

public static void addRule(Rule rule) throws Exception {
AlertingRuleSource sourceObject = DynamicRuleSource.getSourceObject();
sourceObject.addRule(rule);
}