Code Walkthroughs DataStream API

Posted xiaoyon

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Code Walkthroughs DataStream API相关的知识,希望对你有一定的参考价值。

上级: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

 

 

以上是关于Code Walkthroughs DataStream API的主要内容,如果未能解决你的问题,请参考以下文章

日期增加天数--JS Date

uni-app父子通讯,跨页面跨组件通讯

uni-app父子通讯,跨页面跨组件通讯

uni-app父子通讯,跨页面跨组件通讯

flink example运行失败,报错classnotfound

promise