Code Walkthroughs DataStream API

Posted 2021-02-27 xiaoyon

上级：https://www.cnblogs.com/hackerxiaoyon/p/12747387.html

DataStream API

　　DataStreamApi 提供了健壮，有状态的流应用，提供了细力度的控制基于状态和时间，事件驱动系统中可以高级实现。通过DataStreamApi 一步一步的向导我们可以学习到一个有状态的流应用。

What Are You Building ？ 

　　通过信用卡交易监控例子，用一个简单的规则set，让我们可以看到flink是如何实现实时业务操作的。

Prerequisites 前提

　　这段代码的演示需要你具备了java 或者 scala的知识，但是你如果你有其他的编程语言能力也是可以的。

 Help,I’m Stuck! 寻求帮助

 

　　如果你卡住了，可以 求助 https://flink.apache.org/gettinghelp.html 。https://flink.apache.org/community.html#mailing-lists 用户邮件列表是一个活跃快速提供帮助的地方。

 

How to Follow Along 如何跟进

 

　　有一些电脑配置：

　　　　Java 8 或者 11 这里官网没有说别的

　　　　Maven

 

　　然后是构建项目：java 和 scala 各一份

$ mvn archetype:generate \\

    -DarchetypeGroupId=org.apache.flink \\

    -DarchetypeArtifactId=flink-walkthrough-datastream-java \\

    -DarchetypeVersion=1.10.0 \\

    -DgroupId=frauddetection \\

    -DartifactId=frauddetection \\

    -Dversion=0.1 \\

    -Dpackage=spendreport \\

    -DinteractiveMode=false

$ mvn archetype:generate \\

    -DarchetypeGroupId=org.apache.flink \\

    -DarchetypeArtifactId=flink-walkthrough-datastream-scala \\

    -DarchetypeVersion=1.10.0 \\

    -DgroupId=frauddetection \\

    -DartifactId=frauddetection \\

    -Dversion=0.1 \\

    -Dpackage=spendreport \\

    -DinteractiveMode=false

 

 

　　之后代码生成了，我们直接看代码。那么如何跟进的演示代码就这样结束了。

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");

    }}

FraudDetector.java

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);

    }}

FraudDetectionJob.scala

package spendreport

import org.apache.flink.streaming.api.scala._import org.apache.flink.walkthrough.common.sink.AlertSinkimport org.apache.flink.walkthrough.common.entity.Alertimport org.apache.flink.walkthrough.common.entity.Transaction