SparkStreaming和Drools结合的HelloWord版详解大数据

package com.dinpay.bdp.rcp.service; 
import java.math.BigDecimal; 
import java.text.DateFormat; 
import java.text.SimpleDateFormat; 
import java.util.Arrays; 
import java.util.Date; 
import org.apache.commons.lang3.StringUtils; 
import org.apache.log4j.Level; 
import org.apache.log4j.Logger; 
import org.apache.spark.SparkConf; 
import org.apache.spark.api.java.JavaPairRDD; 
import org.apache.spark.api.java.function.FlatMapFunction; 
import org.apache.spark.api.java.function.PairFunction; 
import org.apache.spark.api.java.function.VoidFunction; 
import org.apache.spark.api.java.function.VoidFunction2; 
import org.apache.spark.streaming.Durations; 
import org.apache.spark.streaming.Time; 
import org.apache.spark.streaming.api.java.JavaDStream; 
import org.apache.spark.streaming.api.java.JavaPairDStream; 
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream; 
import org.apache.spark.streaming.api.java.JavaStreamingContext; 
import org.kie.api.KieServices; 
import org.kie.api.runtime.KieContainer; 
import org.kie.api.runtime.KieSession; 
import com.dinpay.bdp.rcp.metaq.MetaQReceiver; 
import com.dinpay.bdp.rcp.streaming.StreamingUtil; 
import com.dinpay.bdp.rcp.util.CodisUtil; 
import com.dinpay.bdp.rcp.util.Constant; 
import com.dinpay.dpp.rcp.po.Order; 
import redis.clients.jedis.Jedis; 
import scala.Tuple2; 
/** 
* 同卡号单日最大交易金额测试  
* @author ll-t150 
* 
*/ 
public class SparkDroolsTest { 
public static Logger logger = Logger.getLogger(SparkDroolsTest.class); 
public static final DateFormat df = new SimpleDateFormat("yyyyMMdd"); 
public static void main(String[] args) { 
String zkConnect=Constant.METAZK; 
String zkRoot="/meta"; 
String topic=Constant.ORDERTOPIC; 
String group=Constant.STREAMGROUP;  
//屏蔽日志 
Logger.getLogger("org.apache.spark").setLevel(Level.OFF); 
logger.info("metaq configuration:"+zkConnect+"--"+topic+"--"+group); 
SparkConf sparkConf = new SparkConf().setAppName("SparkDroolsTest").setMaster("local[2]");   
JavaStreamingContext ssc = new JavaStreamingContext(sparkConf, Durations.seconds(1));  
//从metaq取消息 
JavaReceiverInputDStream<Order> lines = ssc.receiverStream(new MetaQReceiver(zkConnect,zkRoot,topic,group)); 
JavaDStream<Order> words = lines.flatMap(new FlatMapFunction<Order, Order>() { 
@Override 
public Iterable<Order> call(Order order) throws Exception { 
return Arrays.asList(new Order[]{order}); 
} 
}); 
//同卡号单日交易最大次数 统计包括成功和未成功的订单 
JavaPairDStream<String, Integer> cardCntPairs = getCardJavaPair(words); 
save2Codis(cardCntPairs); 
ssc.start(); 
ssc.awaitTermination(); 
ssc.close(); 
} 
@SuppressWarnings({ "unchecked", "serial" }) 
public static <T> JavaPairDStream<String, T>  getCardJavaPair(JavaDStream<Order> words){ 
JavaPairDStream<String, T> pairs = null; 
//次数统计 
pairs = (JavaPairDStream<String, T>) words.mapToPair(new PairFunction<Order, String, Integer>() { 
@Override 
public Tuple2<String, Integer> call(Order order) { 
Jedis jedis = CodisUtil.getJedisPool().getResource(); 
String cardCntkey = order.getSystemId()+"_CNT_"+order.getPayerCardNo()+"_"+df.format(new Date()); 
//拼接key，先到codis里面查找对应的key是否存在，若存在就直接取对应的值，然后取值加1 
String value = jedis.get(cardCntkey); 
if (StringUtils.isEmpty(value)) { 
return new Tuple2<String, Integer>(cardCntkey, 1); 
} else { 
return new Tuple2<String, Integer>(cardCntkey, Integer.parseInt(value) + 1); 
} 
} 
}); 
return pairs; 
} 
/** 
* 将计算出的数据保存到codis中 
* @param pair 
*/ 
@SuppressWarnings("serial") 
public static <T> void save2Codis(JavaPairDStream<String, T> pair) { 
pair.foreachRDD(new VoidFunction2<JavaPairRDD<String,T>,Time>() { 
@Override 
public void call(JavaPairRDD<String, T> rdd, Time time) throws Exception { 
rdd.foreach(new VoidFunction<Tuple2<String,T>>() { 
@Override 
public void call(Tuple2<String, T> tp) throws Exception { 
Jedis jedis = CodisUtil.getJedisPool().getResource(); 
jedis.set(tp._1(), String.valueOf(tp._2())); 
logger.info(tp._1() + ">>>" + tp._2()+",保存到Codis完成！"); 
KieServices kieServices = KieServices.Factory.get(); 
KieContainer kieContainer = kieServices.getKieClasspathContainer(); 
KieSession kieSession = kieContainer.newKieSession("helloworld"); 
ChannAmount objectChannel = new ChannAmount(); 
objectChannel.setAmount(Integer.parseInt(String.valueOf(tp._2()))); 
objectChannel.setChannel(tp._1()); 
kieSession.insert(objectChannel); 
kieSession.fireAllRules(); 
if(jedis !=null){ 
jedis.close(); 
} 
} 
}); 
} 
}); 
} 
}

<?xml version="1.0" encoding="UTF-8"?> 
<kmodule xmlns="http://jboss.org/kie/6.0.0/kmodule"> 
<kbase name="rules" packages="rules"> 
<ksession name="helloworld"/> 
</kbase> 
<kbase name="dtables" packages="dtables"> 
<ksession name="ksession-dtables"/> 
</kbase> 
<kbase name="process" packages="process"> 
<ksession name="ksession-process"/> 
</kbase> 
</kmodule>

package rules; 
import com.dinpay.bdp.rcp.service.ChannAmount; 
//其中m为对象objectChannel 的引用 
rule "channel" 
when 
ChannAmount(amount>2) 
then 
System.out.println("Drools规则实现：该渠道最近5分钟交易金额超过2次 "); 
end