We were using cron in our previous solution, which ran big SQL query once a day and created report from found transactions. We wanted to make this fraud detection process realtime and to know about suspicious situations as fast as possible. That is why we decided to use TouK Nussknacker.<\/p>\n
TouK Nussknacker<\/h2>\n![\"Nussknacker](\"https:\/\/gist.githubusercontent.com\/alien11689\/5d35fdcd91a7304e5eda5123f5bb6b54\/raw\/0c9ddf79642e0349860e527d89e91132f489c491\/nussknacker.png\")
<\/div>\nNussknacker<\/a> is a GUI to the Apache Flink<\/a> developed by TouK<\/a> to help its clients manage various event stream processes. Nussknacker was first deployed in 2016 at one of the Polish mobile network operators. The GUI helps business analysts, marketers and other trained users create and manage processes with a minimal support from developers.<\/p>\nArchitecture<\/h2>\n
Architecture of our solution was based on demo project available in Nussknackera github project<\/a>. It contains:<\/p>\n\nUI<\/code> – allows us to create, test and deploy flink jobs and also read the metrics or logs of them.<\/li>\njob manager<\/code> – dispatches flink jobs to task managers<\/li>\ntask manager<\/code> – runs flink jobs<\/li>\nzookeeper<\/code> – for discovery of task managers and keeping information about state<\/li>\nlogstash<\/code> – harvests the logs from flink tasks, parse them and send to elasticsearch<\/code><\/li>\nelasticsarch<\/code> – keeps logs in index for further analyze<\/li>\nkibana<\/code> – allows for searching in harvested logs<\/li>\ninfluxdb<\/code> – keeps jobs metrics<\/li>\ngrafana<\/code> – allows for searching and viewing jobs metrics<\/li>\nmysql<\/code> – the source of transactions<\/li>\n<\/ul>\nPrepare components<\/h2>\n
Nussknacker assumes some initial effort to create a problem model and components which know how to convert real data into the model, aggregate it and send to external destination.<\/p>\n
The model<\/h3>\n
If we want to create a Nussknacker process, we have to create a model first. The model is a description of data, but also contains all data that might be used in the future or displayed in logs.<\/p>\n
In our case we have created Transaction<\/code> case class in Scala:<\/p>\ncase class Transaction(\r\n id: Long,\r\n amount: BigDecimal,\r\n customerMsisdn: String,\r\n date: LocalDateTime,\r\n externalCustomerId: String,\r\n uid: String,\r\n paymentMethodId: String,\r\n statusId: Long,\r\n status: String,\r\n failureType: String,\r\n failureDetails: String,\r\n drawDownBillerId: Long,\r\n paymentProvider: String,\r\n paymentProviderId: Long,\r\n product: String) extends WithFields {\r\n\r\n val timestamp: Long = date.toInstant(ZoneOffset.UTC).toEpochMilli\r\n\r\n \/\/ ...\r\n}\r\n<\/pre>\nIt is important to create a timestamp field representing when the event occured. Thanks to that aggregates are calculated in valid and deterministic way. This is very important for historical or delayed data.<\/p>\n
Build input<\/h3>\n
The data is kept in the MySQL database, therefore flink needs information about how to fetch it, convert to the model and emit to processing.<\/p>\n
We have created JdbcTransactionSource<\/code> data class:<\/p>\ncase class JdbcTransactionSource(limit: Long, intervalInSeconds: Long, jdbcConfig: JdbcConfig)\r\n extends RichSourceFunction[Transaction]\r\n with Checkpointed[java.lang.Long]\r\n{\r\n \/\/ ...\r\n}\r\n<\/pre>\nThe amount of data to fetch is limited by the source, because we do not want to take all the database records on first start or start after long period of time. What is more, a time interval is set between data fetches. The source has checkpoints stored in the HDFS or in other configured storage with metadata saved in zookeeper to know which records have already been processed.<\/p>\n
The whole job of source class could be summarized by these lines:<\/p>\n
val rs = statement.executeQuery()\r\nwhile (rs.next()) {\r\n val transaction = fromResultSetRow(rs)\r\n sourceContext.collectWithTimestamp(transaction, transaction.timestamp)\r\n lastTransactionId = transaction.id\r\n}\r\n<\/pre>\nBuild sink<\/h3>\n
Sink is a component which knows where to send data after processing. In our case, we want to just log a message with transaction. Generated logs will be sent to elasticsearch by logstash.<\/p>\n
Sink code is quite simple:<\/p>\n
class LoggingSink(messagePrefix: String, loggingLevel: String) extends FlinkSink with LazyLogging with Serializable {\r\n\r\n override def toFlinkFunction: SinkFunction[Any] = new SinkFunction[Any] with Serializable {\r\n override def invoke(in: Any): Unit = {\r\n val message = s\"$messagePrefix - $in\"\r\n loggingLevel match {\r\n case \"error\" => logger.error(message)\r\n case \"warn\" => logger.warn(message)\r\n case default => logger.info(message)\r\n }\r\n }\r\n }\r\n\r\n override def testDataOutput: Option[(Any) => String] = Some(in => s\"[$loggingLevel] $messagePrefix $in\")\r\n}\r\n<\/pre>\n <\/p>\n
Create aggregate function<\/h3>\n
One of the more difficult tasks is to a create our custom aggregate function. In our case, it is counter function, which will count occurences of transactions that meet our conditions. Each aggregate is created for a certain key. In our case, the key is customerMsisdn<\/code>, since it points directly at a single customer. For each customer we want to know how many transactions in a time window were made or found.<\/p>\nWe have created EventsCounter<\/code> class which parses input parameters from configuration and based on the key transforms incoming events and stores count in the eventsCounter<\/code> variable:<\/p>\nclass EventsCounter extends CustomStreamTransformer {\r\n\r\n @MethodToInvoke(returnType = classOf[EventCount])\r\n def execute(@ParamName(\"key\") key: LazyInterpreter[String],\r\n @ParamName(\"length\") length: String) = {\r\n FlinkCustomStreamTransformation((start: DataStream[InterpretationResult]) => {\r\n val lengthInMillis = Duration(length).toMillis\r\n start.keyBy(key.syncInterpretationFunction)\r\n .transform(\"eventsCounter\", new CounterFunction(lengthInMillis))\r\n })\r\n }\r\n}\r\n<\/pre>\n <\/p>\n
\n@MethodToInvoke<\/code> tells Nussknacker that the annotated method should be invoked when process reaches component<\/li>\n@ParamName<\/code> tells Nussknacker that component is configurable via parameters provided in UI<\/li>\n<\/ul>\nCounterFunction<\/code> counts occurences in the time window and stores them as EventCount<\/code> objects which knows when the first occurence of event in this time window was:<\/p>\nclass CounterFunction(lengthInMillis: Long) extends TimestampedEvictableState[Long] {\r\n\r\n override def stateDescriptor =\r\n new ValueStateDescriptor[MultiMap[Long, Long]](\"state\", classOf[MultiMap[Long, Long]])\r\n\r\n override def processElement(element: StreamRecord[InterpretationResult]): Unit = {\r\n setEvictionTimeForCurrentKey(element.getTimestamp + lengthInMillis)\r\n state.update(filterState(element.getTimestamp, lengthInMillis))\r\n\r\n val ir = element.getValue\r\n val eventCount = stateValue.add(element.getTimestamp, 1)\r\n state.update(eventCount)\r\n\r\n val eventsCount = eventCount.map.values.flatten.sum\r\n val smallestTimestamp = eventCount.map.keys.min\r\n output.collect(new StreamRecord[ValueWithContext[Any]](\r\n ValueWithContext(EventCount(count = eventsCount, smallestTimestamp = smallestTimestamp), ir.finalContext), element.getTimestamp)\r\n )\r\n }\r\n}\r\n\r\ncase class EventCount(count: Long, smallestTimestamp: Long)\r\n<\/pre>\nBuild Nussknacker process<\/h2>\n
We have created some components needed for our domain problem. We created a jar library with a class describing process building blocks:<\/p>\n
class FraudDetectorConfigCreator extends ProcessConfigCreator {\r\n val transactionCategory = \"TransactionStream\"\r\n\r\n override def sourceFactories(config: Config): Map[String, WithCategories[SourceFactory[_]]] = {\r\n import net.ceedubs.ficus.Ficus._\r\n import net.ceedubs.ficus.readers.ArbitraryTypeReader._\r\n Map(\r\n \"Transactions\" -> WithCategories(new TransactionSourceFactory(config.as[JdbcConfig](\"jdbc\")), transactionCategory)\r\n )\r\n }\r\n\r\n override def customStreamTransformers(config: Config): Map[String, WithCategories[CustomStreamTransformer]] = Map(\r\n \"eventsCounter\" -> WithCategories(new EventsCounter, transactionCategory)\r\n )\r\n\r\n override def sinkFactories(config: Config): Map[String, WithCategories[SinkFactory]] = Map(\r\n \"LoggingFactory\" -> WithCategories(new LoggingSinkFactory, transactionCategory)\r\n )\r\n\r\n \/\/ other factories that we do not need in our problem\r\n}\r\n<\/pre>\nComponents are available via factories and each factory is created in one or more categories (always TranscationStream<\/code> in our case).<\/p>\nconfiguration<\/h3>\n
We provided some configuration for Nussknacker UI.<\/p>\n
fraudDemo {\r\n timeout: 10s\r\n checkpointInterval: 20s\r\n restartInterval: \"10s\"\r\n processConfigCreatorClass: \"pl.touk.zephyr.frauddetector.FraudDetectorConfigCreator\"\r\n jdbc {\r\n driver: \"com.mysql.jdbc.Driver\"\r\n url: \"jdbc:mysql:\/\/mysql-zephyr:3306\/backofficedev\"\r\n userName: \"backofficedev\"\r\n password: \"...\"\r\n } \r\n defaultValues {\r\n values {\r\n }\r\n }\r\n\r\n}\r\n<\/pre>\nThis configuration will be forwarded to flink with our job.<\/p>\n
Process definition<\/h3>\n
After Nussknacker start the UI is available ‘http:\/\/[host]:8080\/<\/a>‘ and we could start defining processes. We created a new process with category TransactionStrem<\/code>. Using our custom components and also generic components (e. g. filter or split) we designed a fraud detection process.<\/p>\n
<\/div>\nNussknacker<\/a> is a GUI to the Apache Flink<\/a> developed by TouK<\/a> to help its clients manage various event stream processes. Nussknacker was first deployed in 2016 at one of the Polish mobile network operators. The GUI helps business analysts, marketers and other trained users create and manage processes with a minimal support from developers.<\/p>\n Architecture of our solution was based on demo project available in Nussknackera github project<\/a>. It contains:<\/p>\n Nussknacker assumes some initial effort to create a problem model and components which know how to convert real data into the model, aggregate it and send to external destination.<\/p>\n If we want to create a Nussknacker process, we have to create a model first. The model is a description of data, but also contains all data that might be used in the future or displayed in logs.<\/p>\n In our case we have created It is important to create a timestamp field representing when the event occured. Thanks to that aggregates are calculated in valid and deterministic way. This is very important for historical or delayed data.<\/p>\n The data is kept in the MySQL database, therefore flink needs information about how to fetch it, convert to the model and emit to processing.<\/p>\n We have created The amount of data to fetch is limited by the source, because we do not want to take all the database records on first start or start after long period of time. What is more, a time interval is set between data fetches. The source has checkpoints stored in the HDFS or in other configured storage with metadata saved in zookeeper to know which records have already been processed.<\/p>\n The whole job of source class could be summarized by these lines:<\/p>\n Sink is a component which knows where to send data after processing. In our case, we want to just log a message with transaction. Generated logs will be sent to elasticsearch by logstash.<\/p>\n Sink code is quite simple:<\/p>\n <\/p>\n One of the more difficult tasks is to a create our custom aggregate function. In our case, it is counter function, which will count occurences of transactions that meet our conditions. Each aggregate is created for a certain key. In our case, the key is We have created <\/p>\n We have created some components needed for our domain problem. We created a jar library with a class describing process building blocks:<\/p>\n Components are available via factories and each factory is created in one or more categories (always We provided some configuration for Nussknacker UI.<\/p>\n This configuration will be forwarded to flink with our job.<\/p>\n After Nussknacker start the UI is available ‘http:\/\/[host]:8080\/<\/a>‘ and we could start defining processes. We created a new process with category Architecture<\/h2>\n
\n
UI<\/code> – allows us to create, test and deploy flink jobs and also read the metrics or logs of them.<\/li>\njob manager<\/code> – dispatches flink jobs to task managers<\/li>\ntask manager<\/code> – runs flink jobs<\/li>\nzookeeper<\/code> – for discovery of task managers and keeping information about state<\/li>\nlogstash<\/code> – harvests the logs from flink tasks, parse them and send to elasticsearch<\/code><\/li>\nelasticsarch<\/code> – keeps logs in index for further analyze<\/li>\nkibana<\/code> – allows for searching in harvested logs<\/li>\ninfluxdb<\/code> – keeps jobs metrics<\/li>\ngrafana<\/code> – allows for searching and viewing jobs metrics<\/li>\nmysql<\/code> – the source of transactions<\/li>\n<\/ul>\nPrepare components<\/h2>\n
The model<\/h3>\n
Transaction<\/code> case class in Scala:<\/p>\ncase class Transaction(\r\n id: Long,\r\n amount: BigDecimal,\r\n customerMsisdn: String,\r\n date: LocalDateTime,\r\n externalCustomerId: String,\r\n uid: String,\r\n paymentMethodId: String,\r\n statusId: Long,\r\n status: String,\r\n failureType: String,\r\n failureDetails: String,\r\n drawDownBillerId: Long,\r\n paymentProvider: String,\r\n paymentProviderId: Long,\r\n product: String) extends WithFields {\r\n\r\n val timestamp: Long = date.toInstant(ZoneOffset.UTC).toEpochMilli\r\n\r\n \/\/ ...\r\n}\r\n<\/pre>\nBuild input<\/h3>\n
JdbcTransactionSource<\/code> data class:<\/p>\ncase class JdbcTransactionSource(limit: Long, intervalInSeconds: Long, jdbcConfig: JdbcConfig)\r\n extends RichSourceFunction[Transaction]\r\n with Checkpointed[java.lang.Long]\r\n{\r\n \/\/ ...\r\n}\r\n<\/pre>\nval rs = statement.executeQuery()\r\nwhile (rs.next()) {\r\n val transaction = fromResultSetRow(rs)\r\n sourceContext.collectWithTimestamp(transaction, transaction.timestamp)\r\n lastTransactionId = transaction.id\r\n}\r\n<\/pre>\nBuild sink<\/h3>\n
class LoggingSink(messagePrefix: String, loggingLevel: String) extends FlinkSink with LazyLogging with Serializable {\r\n\r\n override def toFlinkFunction: SinkFunction[Any] = new SinkFunction[Any] with Serializable {\r\n override def invoke(in: Any): Unit = {\r\n val message = s\"$messagePrefix - $in\"\r\n loggingLevel match {\r\n case \"error\" => logger.error(message)\r\n case \"warn\" => logger.warn(message)\r\n case default => logger.info(message)\r\n }\r\n }\r\n }\r\n\r\n override def testDataOutput: Option[(Any) => String] = Some(in => s\"[$loggingLevel] $messagePrefix $in\")\r\n}\r\n<\/pre>\nCreate aggregate function<\/h3>\n
customerMsisdn<\/code>, since it points directly at a single customer. For each customer we want to know how many transactions in a time window were made or found.<\/p>\nEventsCounter<\/code> class which parses input parameters from configuration and based on the key transforms incoming events and stores count in the eventsCounter<\/code> variable:<\/p>\nclass EventsCounter extends CustomStreamTransformer {\r\n\r\n @MethodToInvoke(returnType = classOf[EventCount])\r\n def execute(@ParamName(\"key\") key: LazyInterpreter[String],\r\n @ParamName(\"length\") length: String) = {\r\n FlinkCustomStreamTransformation((start: DataStream[InterpretationResult]) => {\r\n val lengthInMillis = Duration(length).toMillis\r\n start.keyBy(key.syncInterpretationFunction)\r\n .transform(\"eventsCounter\", new CounterFunction(lengthInMillis))\r\n })\r\n }\r\n}\r\n<\/pre>\n\n
@MethodToInvoke<\/code> tells Nussknacker that the annotated method should be invoked when process reaches component<\/li>\n@ParamName<\/code> tells Nussknacker that component is configurable via parameters provided in UI<\/li>\n<\/ul>\nCounterFunction<\/code> counts occurences in the time window and stores them as EventCount<\/code> objects which knows when the first occurence of event in this time window was:<\/p>\nclass CounterFunction(lengthInMillis: Long) extends TimestampedEvictableState[Long] {\r\n\r\n override def stateDescriptor =\r\n new ValueStateDescriptor[MultiMap[Long, Long]](\"state\", classOf[MultiMap[Long, Long]])\r\n\r\n override def processElement(element: StreamRecord[InterpretationResult]): Unit = {\r\n setEvictionTimeForCurrentKey(element.getTimestamp + lengthInMillis)\r\n state.update(filterState(element.getTimestamp, lengthInMillis))\r\n\r\n val ir = element.getValue\r\n val eventCount = stateValue.add(element.getTimestamp, 1)\r\n state.update(eventCount)\r\n\r\n val eventsCount = eventCount.map.values.flatten.sum\r\n val smallestTimestamp = eventCount.map.keys.min\r\n output.collect(new StreamRecord[ValueWithContext[Any]](\r\n ValueWithContext(EventCount(count = eventsCount, smallestTimestamp = smallestTimestamp), ir.finalContext), element.getTimestamp)\r\n )\r\n }\r\n}\r\n\r\ncase class EventCount(count: Long, smallestTimestamp: Long)\r\n<\/pre>\nBuild Nussknacker process<\/h2>\n
class FraudDetectorConfigCreator extends ProcessConfigCreator {\r\n val transactionCategory = \"TransactionStream\"\r\n\r\n override def sourceFactories(config: Config): Map[String, WithCategories[SourceFactory[_]]] = {\r\n import net.ceedubs.ficus.Ficus._\r\n import net.ceedubs.ficus.readers.ArbitraryTypeReader._\r\n Map(\r\n \"Transactions\" -> WithCategories(new TransactionSourceFactory(config.as[JdbcConfig](\"jdbc\")), transactionCategory)\r\n )\r\n }\r\n\r\n override def customStreamTransformers(config: Config): Map[String, WithCategories[CustomStreamTransformer]] = Map(\r\n \"eventsCounter\" -> WithCategories(new EventsCounter, transactionCategory)\r\n )\r\n\r\n override def sinkFactories(config: Config): Map[String, WithCategories[SinkFactory]] = Map(\r\n \"LoggingFactory\" -> WithCategories(new LoggingSinkFactory, transactionCategory)\r\n )\r\n\r\n \/\/ other factories that we do not need in our problem\r\n}\r\n<\/pre>\nTranscationStream<\/code> in our case).<\/p>\nconfiguration<\/h3>\n
fraudDemo {\r\n timeout: 10s\r\n checkpointInterval: 20s\r\n restartInterval: \"10s\"\r\n processConfigCreatorClass: \"pl.touk.zephyr.frauddetector.FraudDetectorConfigCreator\"\r\n jdbc {\r\n driver: \"com.mysql.jdbc.Driver\"\r\n url: \"jdbc:mysql:\/\/mysql-zephyr:3306\/backofficedev\"\r\n userName: \"backofficedev\"\r\n password: \"...\"\r\n } \r\n defaultValues {\r\n values {\r\n }\r\n }\r\n\r\n}\r\n<\/pre>\nProcess definition<\/h3>\n
TransactionStrem<\/code>. Using our custom components and also generic components (e. g. filter or split) we designed a fraud detection process.<\/p>\n
![\"process\"](\"https:\/\/gist.githubusercontent.com\/alien11689\/5d35fdcd91a7304e5eda5123f5bb6b54\/raw\/7d4aa8dd2239a08c7f5d55a697e50806beac7d9c\/process.png\")
<\/div>\n![\"process_after_test\"](\"https:\/\/gist.githubusercontent.com\/alien11689\/5d35fdcd91a7304e5eda5123f5bb6b54\/raw\/7d4aa8dd2239a08c7f5d55a697e50806beac7d9c\/processAfterTest.png\")
<\/div>\n![\"metrics\"](\"https:\/\/gist.githubusercontent.com\/alien11689\/5d35fdcd91a7304e5eda5123f5bb6b54\/raw\/7d4aa8dd2239a08c7f5d55a697e50806beac7d9c\/metrics.png\")
<\/div>\n![\"search\"](\"https:\/\/gist.githubusercontent.com\/alien11689\/5d35fdcd91a7304e5eda5123f5bb6b54\/raw\/7d4aa8dd2239a08c7f5d55a697e50806beac7d9c\/search.png\")
<\/div>\n