大数据-玩转数据-Flink营销对账

一、说明

在电商网站中，订单的支付作为直接与营销收入挂钩的一环，在业务流程中非常重要。对于订单而言，为了正确控制业务流程，也为了增加用户的支付意愿，网站一般会设置一个支付失效时间，超过一段时间不支付的订单就会被取消。另外，对于订单的支付，我们还应保证用户支付的正确性，这可以通过第三方支付平台的交易数据来做一个实时对账。

二、思路

对于订单支付事件，用户支付完成其实并不算完，我们还得确认平台账户上是否到账了。而往往这会来自不同的日志信息，所以我们要同时读入两条流的数据来做合并处理。

三、数据准备

订单数据从OrderLog.csv中读取，交易数据从ReceiptLog.csv中读取
JavaBean类的准备

四、代码

package com.lyh.flink06;import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.util.Collector;import java.util.HashMap;
import java.util.Map;public class Project_Order {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(2);SingleOutputStreamOperator<OrderEvent> orderEventString = env.readTextFile("input/OrderLog.csv").map(line -> {String[] data = line.split(",");return new OrderEvent(Long.valueOf(data[0]),data[1],data[2],Long.valueOf(data[3]));}).filter(log -> "pay".equals(log.getEventType()));SingleOutputStreamOperator<TxEvent> txEventString = env.readTextFile("input/ReceiptLog.csv").map(line -> {String[] data = line.split(",");return new TxEvent(data[0],data[1],Long.valueOf(data[2]));});orderEventString.connect(txEventString).keyBy(OrderEvent::getTxId,TxEvent::getTxId).process(new KeyedCoProcessFunction<String, OrderEvent, TxEvent, String>() {Map<String,OrderEvent> OrderEventmap = new HashMap<>();Map<String,TxEvent> TxEventmap = new HashMap<>();@Overridepublic void processElement1(OrderEvent value,Context ctx,Collector<String> out) throws Exception {TxEvent txEvent = TxEventmap.get(ctx.getCurrentKey());if (txEvent != null) {out.collect("订单" + value.getOrderId() + "对账成功");}else {OrderEventmap.put(ctx.getCurrentKey(),value);}}@Overridepublic void processElement2(TxEvent value,Context ctx,Collector<String> out) throws Exception {OrderEvent orderEvent = OrderEventmap.get(ctx.getCurrentKey());if (orderEvent != null) {out.collect("订单" + orderEvent.getOrderId() + "对账成功");}else {TxEventmap.put(ctx.getCurrentKey(),value);}}}).print();env.execute();}
}

五、结果