Rule Execution as Streaming Process with Flink

As explained in the above diagram, rule creator (Desktop) will create JSON based rule and push them to Kafka (rule topic). Event Source will send events to Kafka  (testin topic). Finally Flink will consume both rules and events as streams and process rules based on key (Driver Id). Rules will be stored in Flink as in-memory collection and the rules also can be updated in same manner. Finally out put result will be send to Kafka (testout topic).

Setup Flink

• Download Apache Flink 1.6.3 from below location and extract archive file.

          https://www.apache.org/dyn/closer.lua/flink/flink-1.6.3/flink-1.6.3-bin-scala_2.11.tgz

• Download below dependancy jar files and place in flink-1.6.3/lib folder.

• Configure Flink with flink-1.6.3/conf/flink-conf.yaml .

         Change job-manager and task-manager heap size to much smaller size.

         jobmanager.heap.size: 1g
         taskmanager.heap.size: 2g
         taskmanager.numberOfTaskSlots: 20

• Change State Backend to RocksDB

          Create folder from you home location

         ~$ mkdir -p data/flink/checkpoints

          Edit configuration

          state.backend: rocksdb
          state.checkpoints.dir: file:///home/dhanuka/data/flink/checkpoints

• Start Flink cluster in standalone mode.

           :~/software/flink-1.6.3$ ./bin/start-cluster.sh

Setup Kafka & Zookeeper

• In here I am using docker and docker compose so you can follow below blogspost to install docker & docker-compose in ubuntu.

          http://dhanuka84.blogspot.com/2019/02/install-docker-in-ubuntu.html

• Please checkout below docker project from github.

          https://github.com/dhanuka84/my-docker.git 

• Change IP address to your machine IP address.

         https://github.com/dhanuka84/my-docker/blob/master/kafka/kafka-hazelcast.yml

• Bootup Kafka and Zookeeper with docker-compose

          my-docker/kafka$ sudo docker-compose -f kafka-hazelcast.yml up

• Check docker containers

• Create Kafka Topics

         Download Confluent Platform - https://www.confluent.io/download/

• Got to confluent platform extracted location and run below commands

bin/kafka-topics  --create --zookeeper localhost:2181 --replication-factor 1 --partitions 6 --topic  testin

bin/kafka-topics  --create --zookeeper localhost:2181 --replication-factor 1 --partitions 6 --topic  testout

bin/kafka-topics  --create --zookeeper localhost:2181 --replication-factor 1 --partitions 6 --topic  rule

Create Java based Rule Job

• Checkout the project from github and build the project with maven.

          https://github.com/dhanuka84/stream-analytics
       
          stream-analytics$ mvn clean install -Dmaven.test.skip=true

• FlinkRuleJob.java

          Please note that both rules and events filter by key.

         

          KeyedStream<Event, String> filteredEvents = events.keyBy(Event::getDriverID);

          rules.broadcast().keyBy(Rule::getDriverID)

• FilterFunction.java

          Store rules in ListState using flatMap1 method.

          Execute rule against  each relevant event using flatMap2 method.

• TransformFunction.java

          Finally results transform to JSON string.

• Sample Rule
• Sample Event

• Prepare flat jar to upload, using below command within checked out project home.

stream-analytics$ jar uf target/stream-analytics-0.0.1-SNAPSHOT.jar application.properties consumer.properties producer.properties

• Upload Flink Job Jar file

         copy stream-analytics-0.0.1-SNAPSHOT.jar Flink home folder and run below command

         bin/flink run stream-analytics-0.0.1-SNAPSHOT.jar

Testing

• Run Producer.java class using your favorite IDE. This will generate both events and rules then publish to Kafka.
• To verify from Kafka level use below commands within confluent home location.

Get offset

bin/kafka-run-class kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic testout  --time -1

 Read from Kafka topic

bin/kafka-console-consumer --bootstrap-server localhost:9092 --topic testout

Reference

https://www.ververica.com/blog/bettercloud-dynamic-alerting-apache-flink    

Short Notes - Flink Architecture - Task Execution

Flink within YARN

Application Submission and Component interaction

Application

The application consists of a so-called JobGraph, a logical dataflow graph

Job Manager

Single application control by single Job Manager

For each job client is submitted , the session will spawn its own JobManager.

• All Jobs run under session-user credentials
•  Resource Manager holds on to containers for a certain time
•  Internally, sessions built on the dispatcher component. 
•  TaskManager == YARN Container

Data Flow Graph

Blue Nodes == tasks

Edge == data dependency

Data Exchange Strategies

Forward

Sending data from one task to a receiving task

Broadcast

Send all the data to all parallel tasks

Key-based

Guarantees that data items having the same key will be processed by the same task

Random

Data distribute randomly to tasks

Parallel Task Execution

• C & A operators are data sources while E is a data sink.   

• TaskManager runs only tasks of one application

• Scheduling tasks as slices to slots has the advantage that many tasks are co-located on the TaskManager which means that they can efficiently exchange data without accessing the network.

 

References:

https://www.safaribooksonline.com/library/view/stream-processing-with/9781491974285/

Complete Solution for SQL Based Real Time Streaming Analytics & Machine Learning

 

This hypothetical and high level architecture diagram will explain how we can use OSS technologies more effectively. We are going to discuss about this whole solution under four main subject areas.

1. Distributed Streaming Message Broker (Data Pipeline)
2. Streaming ETL & Streaming Aggregation
3. Time Series Data Storing & ML Processing
4. Dashboard & Notification 

Distributed Streaming Message Broker

We use Kafka as our data pipeline, you can get more details about Kafka from here [1] .
Kafka is proven technology which perform as distributed streaming log. Using it's partitioning technique we can scale up easily. It's perfectly ideal for event driven architecture because it's streaming nature. Kafka make it much more likely that disk access is often sequential and it utilized OS page cache efficiently [2].

Streaming ETL & Streaming Aggregation

Flink will be the stream processing platform that we use for aggregation, ETL and CEP. Flink by-design support stream processing and it's widely use streaming technology at the moment. Flink is based on the DataFlow model which means, Flink is processing the elements as and when they come rather than processing them in micro-batches (which is done by Spark streaming).

Uber AthenaX is a SQL based streaming analytics framework [3]. It's combination of YARN , Calcite & Flink. AthenaX give APIs to monitor, access & administrate life cyclone of Flink Jobs. Also because of Calcite we can write SQL based stream processing applications and run them inside AthenaX easily.

Time Series Data Storing & ML Processing

We use Cassandra as time series data storage. Cassandra architecture is ideal for this purpose because of sequential writing to disk, which will help for fast reading large set of data.

So the time series prepared data we got from streaming analytics platform, will be stored in Cassandra for machine learning.

We will train several ML models using prepared data, then trained models/knowledge base will be stored (as binary) in Cassandra.This way we can dynamically select trained models when we analyze real time data.

Spark slave/worker and Cassandra will reside in same host, and both will be connected by Spark-Cassandra-Connector. For high availability there will be active and stand by Spark master while Zookeeper keep the state of each master.

The main advantage of this approach is  guarantee of data locality between Cassandra node and Spark slave, which will cause for high performance data fetching.

Dashboard & Notification

Finally you need to store analyzed data for later use (Dashboard & trend analysis). Say for example, If we use this solution for predictive analysis, you can save predicted data in ElasticSearch. By using ELK stack, people can create Dashboards easily using analysis data with Kibana.

Also there are lot of other features in ElasticSearch & Kibana such as ES alerts, Trend analysis, Searching and etc...


References:

[1] https://kafka.apache.org/
[2] https://stackoverflow.com/questions/45751641/how-does-kafka-guarantee-sequential-disk-access
[3] https://eng.uber.com/tag/athenax/
[4] https://www.infoq.com/presentations/uber-ml-architecture-models?utm_source=infoqemail&utm_medium=ai-ml-data-eng&utm_campaign=newsletter&utm_content=05152018

Fast Stream Analytics with Kafka & Flink

Steps:

1.Setup Kafka Connect & Kafka

http://dhanuka84.blogspot.com/2017/03/kafka-connect-architecture.html

2. Download Flink & Install

https://ci.apache.org/projects/flink/flink-docs-release-1.2/quickstart/setup_quickstart.html

3. Go to flink folder : cd flink-1.2.0/

4. Start flink in standalone mode

./bin/start-local.sh

5. Start Kafka Read Job

./bin/flink run ~/flink/flink-examples/flink-examples-streaming/target/Kafka.jar --topic metric_topic --bootstrap.servers localhost:9092

6. Watch logs

tail -f log/flink-*-jobmanager-*.out

7. Send JSON to Kafka Connect TCP port.

cat analytics.json | netcat localhost 12345

Source Code: 

https://github.com/dhanuka84/flink/blob/master/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/kafka/ReadFromKafka.java