Tutorial
8 min read

Flink with MLeap

MLOps with Stream Processing

In the big data world, more and more companies are discovering the potential in fast data processing using stream technologies. We don't need to even mention how useful Machine Learning is, as it goes without saying. 

Usually, not a lot of experts master these two domains. So, integrating your streaming architecture with the accomplishments of your data scientist team may not be so easy. 

In this article, I would like to present one of the possible ways to show how this kind of integration is possible. To achieve this, I chose Apache Flink, a streaming engine widely used across the world and MLeap, which is a serialization library for machine learning models for different programming languages. You can find more information about MLeap and how to create a model in Python and run it in Scala in my previous article here

The idée fixe of this paper is to explain how the ML model trained in Python can be used inside the Flink pipeline and provide information about the Flink library we at  GetInData are working on. Check details here.

Motivation for Flink with MLeap

Why serve ML models inside a streaming engine? Why not just reuse some HTTP service that can provide us with predictions or read (predicted and saved data) from DB? 

Here are three arguments: 

  • The need for ultra-low latencies - serving predictions inside streaming jobs is much faster as we can omit network latencies caused by connections to an HTTP server or DB. Not mentioning the fact that the network sometimes fails. 
  • Service decoupling - what if other teams change the REST JSON payload or a  Data Analyst renames the column in the database? Storing models inside Flink jobs is a simple solution. 
  • Costs - nobody likes to talk about them, still they are one of the main factors if projects are to be successful. Keeping additional HTTP servers and DBs + pipelines that compute predictions and save them to DB generates costs. Your management team doesn't like costs, I'm 100% sure about that!  

Of course, storing ML models inside Jobs create other challenges, like: 

  • How do you load the model?
  • How do you make predictions? 
  • My team is using Flink SQL. Is it possible to make predictions using this API? 

To answer these questions, we have created the Flink-MLeap library which you can find here.

Use Case 

Imagine you are a Data Scientist (or maybe you actually are) and would like to use the existing Flink infrastructure to plug your trained ML model and make predictions on streamed data. You're a very busy person and don’t want to waste time on learning totally new technology. It's alright, you're not being lazy! It's only natural, no-one can learn all the different technologies in the world. 

Let's focus on what you know, and how to use this knowledge. You're a “data guy/girl” so you've probably heard about SQL, or maybe even written thousands of queries in this language. Good! When you were reading about Flink you came across this interesting blog post about building Flink jobs with SQL, like this one here

Connecting to the Flink cluster using SQL Client and writing queries to make predictions on streams should be so easy and natural, right?

Usage

With our library, you can effortlessly serve ML models in a stream environment: 

So first we create a stream with features using SQL. In order to do that we use a datagen connector which is a helper connector in Flink, that generates a stream with random values - very useful for the development phase. 

// Create table with features
CREATE TABLE Features (
  feature1 DOUBLE NOT NULL,
  feature2 INT NOT NULL,
  feature3 DOUBLE NOT NULL,
  feature_timestamp TIMESTAMP(3))
WITH ( 
'connector' = 'datagen',
'number-of-rows' = '10',
'fields.feature1.min' = '0.0',
'fields.feature1.max' = '1.0'
)


Next we make predictions based on those features:

// Execute predictions
SELECT
  Predict(feature1, feature2, feature3) as prediction,
  Predictv2(feature1) as prediction2
FROM Features


As you can see, here we are using the SQL user defined function Predict and Predictv2 on our features set. They can take a different number of arguments and types. The names of functions and models they use can be simply defined in the configuration. 

Below you can find more technical aspects of the library, how we built it and examples of how to use and configure it.

FLINK SQL API 

We focused more on Flink SQL API and prepared more utils for this API, so someone who's not too familiar with Flink or feels more comfortable using SQL rather than Java/Scala can effortlessly use the ML model in Flink jobs. 

To this end, we prepared MLeapUDFRegistry. The main purpose of this registry is to register UDFs (Flink user defined functions) that can be later used in SQL queries. To add your UDFs, you can define them inside application.conf like this:

mleap {
	udfRegistry = [
    	{
        	udfName = "Predict"
        	bundlePath = "/mleap-example-1"
        	bundleSource = "file"
    	},
    	{
           udfName = "Predictv2"
        	bundlePath = "/mleap-example-2"
        	bundleSource = "file"
    	}
	]
}

And run  MLeapUDFRegistry.registerFromConfig(config, tableEnv) before running your queries like  we did in these example  applications: FlinkSqlWithMLeap.

...
	val env = StreamExecutionEnvironment.getExecutionEnvironment
	val tableEnv = StreamTableEnvironment.create(env)
 
	// Register UDFs basing on config
	val config = ConfigFactory.load()
	MLeapUDFRegistry.registerFromConfig(config, tableEnv)
 
...

Another thing which can be quite problematic is writing  specific UDFs for each ML Model. This is of course the easiest, but also most time-consuming approach. That's why we defined a very generic MLeapUDF so it can be easily reused for any MLeap bundle. 

Thanks to MLeapUDFRegistry and MLeapUDF, using ML models with SQL goes very smoothly. Just look at the FlinkSqlWithMLeap applications. Anybody who knows SQL and has a ML model can easily use them with Flink.

Bon appetit!

Code revive

Let's have  a quick look at the code. We wrote this project in Scala. It contains two modules: 

  • lib - with library classes
  • example - with usage examples

In the library module we covered two Flink APIs: Streaming and SQL so they can be reused in any job.

Loading ML models

In order to load a model we need to create one first. I will reuse the models presented in the previous article.

Random forest regressors take one float as input and give one float as output.  

To load the models into the Flink jobs, we created BundleLoaders. One is FileBundleLoader which loads bundles from local files. The other is GCSBundleLoader, which can fetch models from Google Cloud Storage Bucket and use them in Flink jobs. 

Streaming API 

In our library we focus more on SQL examples because the audience for that feature is larger. I believe more Data Scientists know SQL than Java. Having said that, the Streaming API was a good starting point for checking if it was even possible to run jobs with MLeap models. 

In MleapMapFunction we presented a way to use MLeap bundles. We loaded the model in the open method. 

case class MleapMapFunction(bundleName: String, bundleLoader: BundleLoader) extends
  RichMapFunction[Double, Double] {
 
  private val LOG = LoggerFactory.getLogger(classOf[MleapMapFunction])
  @transient var transformer: Transformer = _
 
  override def open(parameters: Configuration): Unit = {
	transformer = bundleLoader.loadBundle(bundleName) match {
  	case Failure(exception) => {
    	LOG.error(s"Error while loading bundle: $bundleName", exception)
    	throw BundleLoadProblem(exception)
  	}
  	case Success(value) => value
	}
  }
 
  override def map(value: Double): Double = {
	val dataset = Seq(Row(DenseTensor(Array(value), List(1))))
	val frame = DefaultLeapFrame(transformer.inputSchema, dataset)
	val res = transformer.transform(frame).get.dataset.head(1).asInstanceOf[Double]
	res
  }
}


Then in the map method we make predictions. As you can see, it was a very straightforward solution. 

To test it we implemented a simple Fink job FlinkDatastreamWithMleap:

object FlinkDatastreamWithMleap {
  def main(args: Array[String]): Unit = {
 
	implicit val typeInfo = TypeInformation.of(classOf[StructType])
	val env = StreamExecutionEnvironment.getExecutionEnvironment
 
	val rand: Random = new Random()
 
	val text = env.fromElements(rand.nextDouble(), rand.nextDouble(), rand.nextDouble())
	val bundlePath = getClass.getResource("/mleap-example-1").toString
 
	text.map(MleapMapFunction(bundlePath, FileBundleLoader)).print()
 
	env.execute()
  }
}

Never stop improving 

There is no such system/lib that cannot be enriched or improved. The same thing goes for this library. The main things that we would like to improve on are: 

  • test our generic UDF with more complicated ML models,
  • prepare examples of how to use it with Kubernetes,
  • add support for other ML model serializers similar to MLeap like: PMML, which will provide support for more ML libraries. 

If you're interested in any of these features, please let us know. By doing this we'll know what's more useful for you and can prioritize our work better! Thanks for reading. Don’t forget to check our other blog posts, such as: Online ML Model serving using MLeap and visit our github.

streaming
apache flink
flink
stream processing
ML Model
MLeap
ML
22 February 2022

Want more? Check our articles

5apacheobszar roboczy 1 4
Tutorial

Real-time ingestion to Iceberg with Kafka Connect - Apache Iceberg Sink

What is Apache Iceberg? Apache Iceberg is an open table format for huge analytics datasets which can be used with commonly-used big data processing…

Read more
getindata adam goscicki terraform cloud infrastructure notext
Tutorial

Terraform your Cloud Infrastructure

So, you have an existing infrastructure in the cloud and want to wrap it up as code in a new, shiny IaC style? Splendid! Oh… it’s spanning through two…

Read more
getindata amundsen feast machine learining notext
Tutorial

Machine Learning Features discovery with Feast and Amundsen

One of the main challenges of today's Machine Learning initiatives is the need for a centralized store of high-quality data that can be reused by Data…

Read more
blog7

5 main data-related trends to be covered at Big Data Tech Warsaw 2021 Part II

Trend 4. Larger clouds over the Big Data landscape  A decade ago,  only a few companies ran their Big Data infrastructure and pipelines in the public…

Read more
geospatial analytics hadoop
Use-cases/Project

Geospatial analytics on Hadoop

A few months ago I was working on a project with a lot of geospatial data. Data was stored in HDFS, easily accessible through Hive. One of the tasks…

Read more
1 RsDrT5xOpdAcpehomqlOPg
Big Data Event

2³ Reasons To Speak at Big Data Tech Warsaw 2020 (February 27th, 2020)

Big Data Technology Warsaw Summit 2020 is fast approaching. This will be 6th edition of the conference that is jointly organised by Evention and…

Read more

Contact us

Interested in our solutions?
Contact us!

Together, we will select the best Big Data solutions for your organization and build a project that will have a real impact on your organization.

The administrator of your personal data is GetInData Sp. z o.o. Sp.k with its registered seat in Warsaw (02-508), 39/20 Pulawska St. Your data is processed for the purpose of provision of electronic services in accordance with the  Terms & Conditions. For more information on personal data processing and your rights please see  Privacy Policy.

By submitting this form, you agree to our Terms & Conditions and Privacy Policy