Fraud Detection with the DataStream API  #

FraudDetectionJob.java

package spendreport; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.walkthrough.common.sink.AlertSink; import org.apache.flink.walkthrough.common.entity.Alert; import org.apache.flink.walkthrough.common.entity.Transaction; import org.apache.flink.walkthrough.common.source.TransactionSource; public class FraudDetectionJob {     public static void main(String[] args) throws Exception {         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();         DataStream<Transaction> transactions = env             .addSource(new TransactionSource())             .name("transactions");                  DataStream<Alert> alerts = transactions             .keyBy(Transaction::getAccountId)             .process(new FraudDetector())             .name("fraud-detector");         alerts             .addSink(new AlertSink())             .name("send-alerts");         env.execute("Fraud Detection");     } }

package spendreport; import org.apache.flink.streaming.api.functions.KeyedProcessFunction; import org.apache.flink.util.Collector; import org.apache.flink.walkthrough.common.entity.Alert; import org.apache.flink.walkthrough.common.entity.Transaction; public class FraudDetector extends KeyedProcessFunction<Long, Transaction, Alert> {     private static final long serialVersionUID = 1L;     private static final double SMALL_AMOUNT = 1.00;     private static final double LARGE_AMOUNT = 500.00;     private static final long ONE_MINUTE = 60 * 1000;     @Override     public void processElement(             Transaction transaction,             Context context,             Collector<Alert> collector) throws Exception {         Alert alert = new Alert();         alert.setId(transaction.getAccountId());         collector.collect(alert);     } }

StreamExecutionEnvironment env = StreamExecutionEnvironment . getExecutionEnvironment ();

DataStream<Transaction> transactions = env     .addSource(new TransactionSource())     .name("transactions");