flinkml/flink-jpmml

flink-jpmml is a fresh-made library for dynamic real time machine learning predictions built on top of PMML standard models and Apache Flink streaming engine

flink-jpmml's Issues

Introducing updated flink-jpmml examples

Refactoring ControlStream management

Unresolved Dependencies io.radicalbit#flink-streaming-spec_2.11;0.3.0: not found

Hi, I cloned the repo, release-0.6.2 branch and imported the project to intellij, the SBT build fails both.

[warn] 	::::::::::::::::::::::::::::::::::::::::::::::
[warn] 	::          UNRESOLVED DEPENDENCIES         ::
[warn] 	::::::::::::::::::::::::::::::::::::::::::::::
[warn] 	:: io.radicalbit#flink-streaming-spec_2.11;0.1.0: not found
[warn] 	::::::::::::::::::::::::::::::::::::::::::::::
[warn]
[warn] 	Note: Unresolved dependencies path:
[warn] 		io.radicalbit:flink-streaming-spec_2.11:0.1.0 (/Users/harishsridhar/Documents/Temp/flink-jpmml/flink-jpmml/build.sbt#L64)
[warn] 		  +- io.radicalbit:flink-jpmml-scala_2.11:0.6.2

Am I missing something?

Regards,
Harish.

Updating Flink to 1.3.1

Quick evaluate method should have a different name in respect of the traditional one

Integrate Travis CI

Integrate Travis CI https://travis-ci.org/

Releasing 0.5.1 version

Updating README to version 0.6

Move all the exceptions in a separate package

Do it after #26

Can't load models in a Flink Cluster setup

Hi,

I built a sample flink application that consumes events from Kafka Topic and evaluates the incoming events using a model present in the file system using ModelReader

val prediction = appsDataStream.evaluate(modelReader) {
      //Iris data and modelReader instance
      case (event, model) =>
        val vectorized = event.toDenseVector
        val prediction = model.predict(vectorized, Some(0.0))

        (event, prediction)
    }

This works on a flink local cluster. However this doesn't work when I deploy the applicatoin on a flink cluster with job manager and task manager running in separate nodes.

The reason for this not working is the PMML model along with application is packaged as a fat jar and uploaded to the job manager. The Job manager sends model evaluate task to the task managers and the task managers don't have the PMML model file available.

Currently flink-jpmml implementation only allows the path of the pmml model file to be provided as input and the library implements the model upload using the path. This doesn't work in cluster mode.
We may need additional implementation where by we can provide the whole PMML model file InputStream using which the library can upload the PMML models.

Regards,
Harish.

Employing flink-spec as a separate library

We should investigate if it is possible right now either we need some extra-work on the former library. @maocorte perhaps can answer to it.

Fixing flink-jpmml-examples README file

Commands examples are not correct. On -c option we should report both entrypoint class and the path to .jar file.

Override `toString` method in `ModelId`

In order to return

Updating flink-jpmml project dependencies

The assignee should double-check if the behavior of the api has changed somehow.

Adding Changelog file

Bump master to 0.6.0-SNAPSHOT

Bump master to 0.6.0-SNAPSHOT version and update the README file accordingly

Handle additional dependencies

Hi,

In the flink-jpmml documentation, it is mentioned that the dependency to be added to projects is as follows.

"io.radicalbit" %% "flink-jpmml-scala" % "0.6.3"

But when I tried to build a simple example using the IRIS datasource, I figured that there needs to be additional dependencies for the project such as the following

"org.jpmml" % "pmml-evaluator" % "1.3.9",
 "org.glassfish.jaxb" % "jaxb-runtime" % "2.3.2"

I wanted to suggest to either add these additional manadatory dependencies to the library"io.radicalbit" %% "flink-jpmml-scala" % "0.6.3" or mention it alteast in README as a note under dependency section.

Regards,
Harish.

Enrich Exception description in README file

Update jpmml version to 1.4.0

Current release version is 1.3.9

Exposing summary method on PmmlModel

The object ModelEvaluator has a getter summary which describes the model in its form. We should expose this method properly.

Avoiding boilerplate to Vector handler Type Class

Publish the 0.6.1 version on a public repo

Publish the 0.6.1 version on a public repo. Evaluate to publish on http://central.sonatype.org/pages/ossrh-guide.html

Update flink 1.3 to latest minor version

Extending evaluator ADT with Empty predictions

Remove requirement for `streaming-scala` Flink library on ServingMessage import

If the user wants to import ServingMessage ADT components from an external project, he's forced to import non-necessary flink dependencies. This is due to the package object scala related to the same package, in which the ADT is located.

This should be fixed by separating the ADT to that logic.

Ensure tests execution during sbt assembly

Evaluate removing hadoop dependencies

Since the lib relies to Flink underlying FS management we should not test whenever a PMML model is stored on top of HDFS, either any hadoop implementation. So my proposal is to remove the former dependencies; moreover, it will boost up performances.

Integer Type Target value not supported

Hi,

The model I used to test the flink-jpmml library had the target value as Integer type.
Following is the target field from the PMML model I used.

<DataDictionary>
		<DataField name="Label" optype="categorical" dataType="integer">
			<Value value="0"/>
			<Value value="1"/>

This resulted in EmptyPrediction for all my inputs.
When I debugged, I found that this is because the integer type target field is not supported in library implementation.

Following is the place where it breaks.

pmmlModel.scala method extractTarget

private[api] def extractTarget(evaluationResult: java.util.Map[FieldName, _]): Double = {
    val targets = extractTargetFields(evaluationResult)

    targets.headOption.flatMap {
      case (_, target) => extractTargetValue(target)
    } getOrElse (throw new JPMMLExtractionException("Target value is null."))

pipeline.scala method extractTargetValue

@throws(classOf[ClassCastException])
  protected def extractTargetValue(target: Any): Option[Double] = target match {
    case s: String => Some(s.toDouble)
    case d: Double => Some(d)
    case _ => None
  }

when I change the target value type to be double in the pmml model I get expected predictions.

Can you check this?

Regards,
Harish.

README.md - Library imports

Good morning.

I write this issue because I've had some trouble getting the library to work. The thing is, in the main README file, there is this piece of code:

// This will be all that you need
import io.radicalbit.flink.pmml.scala._

import org.apache.flink.ml.math.Vector
import org.apache.flink.streaming.api.scala._

object FlinkJpmmlExample {
  def main(args: Array[String]): Unit = {

    // your model can reside in any Flink supported backend
    val pathToPmml = "/even/to/distributed/systems"

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val events = env.yourInputStream

    //  a lazy reader implementation
    val reader = ModelReader(pathToPmml)

However, I couldn't get this piece of code to work. After searching through the repo, I found that an additional import was required:

import io.radicalbit.flink.pmml.scala.api.reader.ModelReader

Could this import be added to the README example? I know it's already in the examples, but not having it in the main README file could be misleading.

For reference, I tried this on an Ubuntu 18.04 VM, working with Flink 1.9.0 and Scala 2.11.12, compiled with sbt 1.0.

Releasing flink-jpmml 0.6.0

This is blocked by #19 #26 #27 #28 #29

Add maven libraries version badges

Add maven libraries version badges to show on README the last published versions for both scala 2.10 and 2.11

Update flink version

By now we have 1.4.2

Introducing Dynamic Model Evaluation