comment utiliser les baves dans la topologie des tempêtes

Question

Maintenant, je veux utiliser Drools dans un blot, cela fonctionne normalement dans LocalCluster, mais quand je l'ai mis dans le cluster de production, il a une erreur. Le transfert est:

public class DealLostBolt extends BaseRichBolt { 

     private static final long serialVersionUID = 1L; 

     private static final Logger LOGGER = LoggerFactory.getLogger("DEAL_LOST_BOLT"); 

     private OutputCollector collector; 

     private KieSession kieSession; 

     private FactHandle factHandle; 

     @Override 
     public void execute(Tuple input) { 
     // 获取数据 
     String sentence = (String) input.getValue(0); 
     LOGGER.info("DealLostBolt获取到的数据：" + sentence); 

     // 数据转换 
     PutDataPoint dataPoint = Json.fromJson(PutDataPoint.class, sentence); 

     KieServices ks = KieServices.Factory.get(); 
     KieContainer kieContainer = ks.getKieClasspathContainer(); 
     kieSession = kieContainer.newKieSession("all-rule"); 
     kieSession.getAgenda().getAgendaGroup("deal-lost").setFocus(); 

     factHandle = kieSession.insert(dataPoint); 
     kieSession.fireAllRules(); 
     kieSession.delete(factHandle); 

     collector.emit(new Values(sentence)); 
     } 

     @Override 
     public void declareOutputFields(OutputFieldsDeclarer declarer) { 
     declarer.declare(new Fields("value")); 

     } 

     @Override 
     public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) { 
     this.collector = collector; 
     } 

    } 

je les documents officiels pour créer le kiesession. Les erros est:

java.lang.RuntimeException: java.lang.NullPointerException 
    at org.apache.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:495) ~[storm-core-1.1.1.jar:1.1.1] 
    at org.apache.storm.utils.DisruptorQueue.consumeBatchWhenAvailable(DisruptorQueue.java:460) ~[storm-core-1.1.1.jar:1.1.1] 
    at org.apache.storm.disruptor$consume_batch_when_available.invoke(disruptor.clj:73) ~[storm-core-1.1.1.jar:1.1.1] 
    at org.apache.storm.daemon.executor$fn__5030$fn__5043$fn__5096.invoke(executor.clj:848) ~[storm-core-1.1.1.jar:1.1.1] 
    at org.apache.storm.util$async_loop$fn__557.invoke(util.clj:484) [storm-core-1.1.1.jar:1.1.1] 
    at clojure.lang.AFn.run(AFn.java:22) [clojure-1.7.0.jar:?] 
    at java.lang.Thread.run(Thread.java:745) [?:1.8.0_121] 
    Caused by: java.lang.NullPointerException 
    at org.kie.internal.io.ResourceFactory.newByteArrayResource(ResourceFactory.java:66) ~[se-storm-0.0.1-SNAPSHOT-jar-with-dependencies.jar:?] 
    at org.drools.compiler.kie.builder.impl.AbstractKieModule.getResource(AbstractKieModule.java:299) ~[se-storm-0.0.1-SNAPSHOT-jar-with-dependencies.jar:?] 
    at org.drools.compiler.kie.builder.impl.AbstractKieModule.addResourceToCompiler(AbstractKieModule.java:264) ~[se-storm-0.0.1-SNAPSHOT-jar-with-dependencies.jar:?] 
    at org.drools.compiler.kie.builder.impl.AbstractKieModule.addResourceToCompiler(AbstractKieModule.java:259) ~[se-storm-0.0.1-SNAPSHOT-jar-with-dependencies.jar:?] 
    at org.drools.compiler.kie.builder.impl.AbstractKieProject.buildKnowledgePackages(AbstractKieProject.java:228) ~[se-storm-0.0.1-SNAPSHOT-jar-with-dependencies.jar:?] 
    at org.drools.compiler.kie.builder.impl.AbstractKieModule.createKieBase(AbstractKieModule.java:206) ~[se-storm-0.0.1-SNAPSHOT-jar-with-dependencies.jar:?] 
    at org.drools.compiler.kie.builder.impl.KieContainerImpl.createKieBase(KieContainerImpl.java:584) ~[se-storm-0.0.1-SNAPSHOT-jar-with-dependencies.jar:?] 
    at org.drools.compiler.kie.builder.impl.KieContainerImpl.getKieBase(KieContainerImpl.java:552) ~[se-storm-0.0.1-SNAPSHOT-jar-with-dependencies.jar:?] 
    at org.drools.compiler.kie.builder.impl.KieContainerImpl.newKieSession(KieContainerImpl.java:680) ~[se-storm-0.0.1-SNAPSHOT-jar-with-dependencies.jar:?] 
    at org.drools.compiler.kie.builder.impl.KieContainerImpl.newKieSession(KieContainerImpl.java:648) ~[se-storm-0.0.1-SNAPSHOT-jar-with-dependencies.jar:?] 
    at cn.ennwifi.storm.bolt.DealLostBolt.execute(DealLostBolt.java:52) ~[se-storm-0.0.1-SNAPSHOT-jar-with-dependencies.jar:?] 
    at org.apache.storm.daemon.executor$fn__5030$tuple_action_fn__5032.invoke(executor.clj:729) ~[storm-core-1.1.1.jar:1.1.1] 
    at org.apache.storm.daemon.executor$mk_task_receiver$fn__4951.invoke(executor.clj:461) ~[storm-core-1.1.1.jar:1.1.1] 
    at org.apache.storm.disruptor$clojure_handler$reify__4465.onEvent(disruptor.clj:40) ~[storm-core-1.1.1.jar:1.1.1] 
    at org.apache.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:482) ~[storm-core-1.1.1.jar:1.1.1] 
    ... 6 more 

Perhapse quelque chose n'initialisation pas. Mais je crée un nouveau kieservice quand blot execute. Quelqu'un pourrait-il m'aider?

Merci!

Answer 1

J'ai eu un problème similaire en utilisant Drools avec JMH en pot ombré. Drools utilise une approche ServiceRegistry. Cela signifie que les bibliothèques Drools (drools-compiler, kie-ci, drools-decisiontables, ...) contiennent un même fichier de propriétés nommé, qui indique l'implémentation pour une interface qu'ils offrent.

Le plugin jar ombré aplatit généralement les dépendances (transitives) dans un pot. Pour les fichiers qui existent plusieurs fois, cela signifie généralement que l'un d'eux est sélectionné s'il n'est pas spécifié autrement. Pour les propriétés ServiceRegistry, nous devons combiner tous les fichiers. Habituellement, cela se fait via le ServicesResourceTransformer. Ce Transformer gère les fichiers dans META-INF/services/, mais le fichier correspondant à Drools est META-INF/kie.conf. Mon problème avec JMH pourrait être résolu avec l'AppendingTransformer:

<transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer"> 
    <resource>META-INF/kie.conf</resource> 
</transformer> 

Je ne suis pas un expert de la tempête, mais le Starter suggère qu'il fait également l'utilisation du plug-in de l'ombre. Je suppose que vous exécutez votre cluster local à partir de l'IDE - qui n'utilise pas le pot ombré.