pragmaticminds/crunch

Industrial data evaluation engine

crunch's Issues

Create initial wiki informations

Topics about the background, the creators, the aims...

SCM repository in POM points to wrong repo

the scm tag points to scm:git:https://github.com/JulianFeinauer/crunch.git,
it should point to the pragmatic minds repo

Create Example module

With code examples and some README documentation.

Documentation

While the readme talks about the the differences in the goal of this project from other streaming data processing frameworks, there is no documentation about the design, the philosophy, or the implementation of crunch.

Major Performance Improvement

From my analysis (see #10) it looks like the main reason for our massive perofrmance degradation is that we use HashMap<String, Object> for the transport of values. But we have too many operations for each record which adds up and leads to several Millions of operations for a mediocre sized batch of messages (140k messages lead to 80 Mio hash calculations on String).

So I suggest to rewrite the value handling internally (not externally) and this should lead to a massive performance gain (I expect x10 - x100 speedup).
There are different ideas I have and I'll try to document it.

But I wanted to open an Issue here to track work on this topic.

Performance Issues for streams with many messages

Currently, we observe a performance problem when processing many messages in a Stream with man evaluation functions or with evaluation functions that "look" at many signals.
I already did some analysis in the branch "performance-test".

The class TestCrunchPerformanceNT contains the main to run the tests (and also a setup to run it with JMH with the IntelliJ Plugin).

The file referenced there is 30MB large and thus not in the repo but is shared in the crunch-oss channel in Slack (or downloaded here: https://pragmaticminds.slack.com/files/U1Q8RMFDL/FFEAR60TS/sample2.txt)

Support for dataformats missing!

Missing is Float, Short, Byte... propably more?

Reported by @timbo2k.

NullPointerException in org.pragmaticminds.crunch.api.trigger.extractor.ChannelMapExtractor.extract(ChannelMapExtractor.java:105)

This Exception occures when an ChannelMapExtractor tries to extract channels, that are not present.

Exception printout:

10:48:45.700 [main] WARN org.pragmaticminds.crunch.execution.CrunchExecutor - Unable to wait for execution of pipeline.
java.util.concurrent.ExecutionException: java.lang.NullPointerException
at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357) ~[?:1.8.0_201]
at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1895) ~[?:1.8.0_201]
at scala.concurrent.java8.FuturesConvertersImpl$CF.super$get(FutureConvertersImpl.scala:82) ~[scala-java8-compat_2.12-0.8.0.jar:?]
at scala.concurrent.java8.FuturesConvertersImpl$CF.$anonfun$get$1(FutureConvertersImpl.scala:82) ~[scala-java8-compat_2.12-0.8.0.jar:?]
at scala.concurrent.BlockContext$DefaultBlockContext$.blockOn(BlockContext.scala:53) ~[scala-library-2.12.6.jar:?]
at scala.concurrent.package$.blocking(package.scala:142) ~[scala-library-2.12.6.jar:?]
at scala.concurrent.java8.FuturesConvertersImpl$CF.get(FutureConvertersImpl.scala:82) ~[scala-java8-compat_2.12-0.8.0.jar:?]
at org.pragmaticminds.crunch.execution.CrunchExecutor.runWithSink(CrunchExecutor.java:101) [crunch-core-0.3.0.jar:0.3.0]
at org.pragmaticminds.crunch.execution.CrunchExecutor.run(CrunchExecutor.java:75) [crunch-core-0.3.0.jar:0.3.0]
at org.pragmaticindustries.inacore.standalone.Application.main(Application.java:29) [classes/:?]
Caused by: java.lang.NullPointerException
at java.util.HashMap.merge(HashMap.java:1225) ~[?:1.8.0_201]
at java.util.stream.Collectors.lambda$toMap$58(Collectors.java:1320) ~[?:1.8.0_201]
at java.util.stream.ReduceOps$3ReducingSink.accept(ReduceOps.java:169) ~[?:1.8.0_201]
at java.util.HashMap$EntrySpliterator.forEachRemaining(HashMap.java:1699) ~[?:1.8.0_201]
at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481) ~[?:1.8.0_201]
at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471) ~[?:1.8.0_201]
at java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:708) ~[?:1.8.0_201]
at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234) ~[?:1.8.0_201]
at java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:499) ~[?:1.8.0_201]
at org.pragmaticminds.crunch.api.trigger.extractor.ChannelMapExtractor.extract(ChannelMapExtractor.java:105) ~[crunch-api-0.3.0.jar:0.3.0]
at org.pragmaticminds.crunch.api.trigger.handler.ExtractorTriggerHandler.lambda$handle$0(ExtractorTriggerHandler.java:99) ~[crunch-api-0.3.0.jar:0.3.0]
at java.util.stream.ReferencePipeline$7$1.accept(ReferencePipeline.java:267) ~[?:1.8.0_201]
at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1382) ~[?:1.8.0_201]
at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481) ~[?:1.8.0_201]
at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471) ~[?:1.8.0_201]
at java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:708) ~[?:1.8.0_201]
at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234) ~[?:1.8.0_201]
at java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:499) ~[?:1.8.0_201]
at org.pragmaticminds.crunch.api.trigger.handler.ExtractorTriggerHandler.handle(ExtractorTriggerHandler.java:104) ~[crunch-api-0.3.0.jar:0.3.0]
at org.pragmaticminds.crunch.api.trigger.TriggerEvaluationFunction.eval(TriggerEvaluationFunction.java:152) ~[crunch-api-0.3.0.jar:0.3.0]
at org.pragmaticminds.crunch.execution.GraphFactory.lambda$toFlow$50b6d5dc$1(GraphFactory.java:209) ~[crunch-core-0.3.0.jar:0.3.0]
at akka.stream.javadsl.Flow.$anonfun$map$1(Flow.scala:484) ~[akka-stream_2.12-2.5.16.jar:2.5.16]
at akka.stream.impl.fusing.Map$$anon$9.onPush(Ops.scala:51) ~[akka-stream_2.12-2.5.16.jar:2.5.16]
at akka.stream.impl.fusing.GraphInterpreter.processPush(GraphInterpreter.scala:519) ~[akka-stream_2.12-2.5.16.jar:2.5.16]
at akka.stream.impl.fusing.GraphInterpreter.execute(GraphInterpreter.scala:411) ~[akka-stream_2.12-2.5.16.jar:2.5.16]
at akka.stream.impl.fusing.GraphInterpreterShell.runBatch(ActorGraphInterpreter.scala:588) ~[akka-stream_2.12-2.5.16.jar:2.5.16]
at akka.stream.impl.fusing.GraphInterpreterShell$AsyncInput.execute(ActorGraphInterpreter.scala:472) ~[akka-stream_2.12-2.5.16.jar:2.5.16]
at akka.stream.impl.fusing.GraphInterpreterShell.processEvent(ActorGraphInterpreter.scala:563) ~[akka-stream_2.12-2.5.16.jar:2.5.16]
at akka.stream.impl.fusing.ActorGraphInterpreter.akka$stream$impl$fusing$ActorGraphInterpreter$$processEvent(ActorGraphInterpreter.scala:745) ~[akka-stream_2.12-2.5.16.jar:2.5.16]
at akka.stream.impl.fusing.ActorGraphInterpreter$$anonfun$receive$1.applyOrElse(ActorGraphInterpreter.scala:760) ~[akka-stream_2.12-2.5.16.jar:2.5.16]
at akka.actor.Actor.aroundReceive(Actor.scala:517) ~[akka-actor_2.12-2.5.16.jar:2.5.16]
at akka.actor.Actor.aroundReceive$(Actor.scala:515) ~[akka-actor_2.12-2.5.16.jar:2.5.16]
at akka.stream.impl.fusing.ActorGraphInterpreter.aroundReceive(ActorGraphInterpreter.scala:670) ~[akka-stream_2.12-2.5.16.jar:2.5.16]
at akka.actor.ActorCell.receiveMessage(ActorCell.scala:588) ~[akka-actor_2.12-2.5.16.jar:2.5.16]
at akka.actor.ActorCell.invoke(ActorCell.scala:557) ~[akka-actor_2.12-2.5.16.jar:2.5.16]
at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:258) ~[akka-actor_2.12-2.5.16.jar:2.5.16]
at akka.dispatch.Mailbox.run(Mailbox.scala:225) ~[akka-actor_2.12-2.5.16.jar:2.5.16]
at akka.dispatch.Mailbox.exec(Mailbox.scala:235) ~[akka-actor_2.12-2.5.16.jar:2.5.16]
at akka.dispatch.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260) ~[akka-actor_2.12-2.5.16.jar:2.5.16]
at akka.dispatch.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339) ~[akka-actor_2.12-2.5.16.jar:2.5.16]
at akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979) ~[akka-actor_2.12-2.5.16.jar:2.5.16]
at akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107) ~[akka-actor_2.12-2.5.16.jar:2.5.16]
10:48:48.155 [Thread-100] INFO org.springframework.boot.context.embedded.AnnotationConfigEmbeddedWebApplicationContext - Closing org.springframework.boot.context.embedded.AnnotationConfigEmbeddedWebApplicationContext@550dbc7a: startup date [Thu Jan 24 10:47:10 CET 2019]; root of context hierarchy

Refactoring / New API

Work is currently already in progress, see https://github.com/pragmaticminds/crunch/tree/feature/tags-and-improved-api

Complete Information in poms

Like Homepages... authors... such things.
@chrisdutz could you help here?

Fix vulnerabilites

Several vulnerabilies are present (akka!!) and others reported by dependency plugin.
This has to be fixed and tested.

Reported by @chrisdutz