我正在尝试使用 Spark Streaming 将 Drools Engine (Drools Fusion) 应用于流数据。
但是我无法反序列化 JavaDStream 类型的变量来触发规则。
它显示 JavaDstream 和 Interger 之间的比较错误。请建议如何解决这个问题。下面是代码
主班
SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("Spark Streaming");
JavaStreamingContext jssc = new JavaStreamingContext(conf, new Duration(10000));
JavaReceiverInputDStream<String> lines = jssc.socketTextStream("localhost", 8089);
class splitFunc implements Function<String, Long> {
public Long call(String x) { return Long.parseLong(x); }
}
JavaDStream<Long> in_amount = lines.map(new splitFunc());
insertEvent(entryPointStoreOne,new Sale(),in_amount);
Sale.java - 事实
import org.apache.spark.streaming.api.java.JavaDStream;
public class Sale {
private JavaDStream<Long> amount;
public Sale (){ }
public void setAmount(JavaDStream<Long> amount)
{
this.amount=amount;
}
public JavaDStream<Long> getAmount()
{
return amount;
}
}
Drools 规则文件
declare Sale
@role(event)
end
rule "Store One - Has Passed its Sales Record"
when
$amount : Sale( amount > 1000 ) - FAILING HERE
from entry-point StoreOne
then
System.out.println("Store One has passed its Sales Records");
end