pragmaticminds/crunch

Industrial data evaluation engine

Readme

Welcome to CRUNCH an industrial streaming data analysis framework built by pragmatic minds GmbH.

About

When dealing with data streams from industrial applications, e.g., machines there are often times very similar questions and processing steps necessary. Think, e.g., of filtering or joining data from different sources. Futhermore, more complex tasks are differentiation of signals (to monitor changes) or even application of Fourrier transformation (or related wavelet transformation). These signals can then be analysed with regards to a set of built-in functions or custom functions. What all of the functions have in common is, that you always have the temporal context of a datapoint. Thus, it is easy to ask

When did this bit change from false to true?

or things like

When is the steepness of this signal larger than ... for more than ... seconds

or

Emit an event each time when the signal is above ...

What makes CRUNCH different from other Frameworks like Apache Flink, Apache Spark, Akka Streams, ...

There are many open source frameworks for stream processing. The main difference between them and CRUNCH is that CRUNCH is very focused about it's application in signal processing and related tasks and no general streaming framework. Futhermore, as this kind of analysis is often done on the edge CRUNCH is not very focused on scaling and fault tolerance in specific situations as this is not (that) relevant for edge devices.

Further Information

Please see on our homepage here: https://pragmaticminds.github.io/crunch/

Developers

To generate the Homepage, just call mvn clean site site:stage scm-publish:publish-scm.

Contact

Obviuously, this readme is still beeing populated and we are still setting up our infrastructure (after open sourcing CRUNCH). So if you have any questions please feel free to ask one of the commiters or write an email to Julian.

crunch's Issues

Support for dataformats missing!

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

Reported by @timbo2k.

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

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.

Fix vulnerabilites

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

Reported by @chrisdutz

Complete Information in poms

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

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

Create Example module

With code examples and some README documentation.

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)

Refactoring / New API

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

Create initial wiki informations

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