文章目录

• 0.数据获取
• 1.需求概要
• 2.数据标准
• • 2.1.输入数据格式
  • 2.2.输出数据格式
  • 2.3.数据主键及关系
• 3.表详细
• • 1.order_info
  • 2.bill_info
  • 3.bill_item
  • 4.ord_pay
  • 5.ord_pay_log
  • 6.pay_method
  • 7.pay_bank_card
• 4.开发工具、语言版本、工具版本
• 5.赛题解答
• • 数据分流
  • 任务1
  • 任务2
  • 任务3
  • 小结
• 附录

0.数据获取

链接: https://pan.baidu.com/s/1NlmbN1cX4K9jYbYQm-IJUw 提取码: qw12
本项目适合有一定flink经验的同学学习或者巩固基础、提升使用。

1.需求概要

在实际的市场交易中为了防范银行卡信息泄露，需要风控系统对存在所有用户操作进行监控，对存在潜在风险的行为进行及时告警。以下是具体风控规则：
规则1：单个用户在一天内，付款超过1万元
源数据：order_info
规则2：在绑卡操作中发现2个不同用户绑定了同一张银行卡；
源数据：pay_method，pay_bank_card
规则3：在付款过程中，发现单个用户在5分钟内使用了10张以上的银行卡付款；
源数据：全部7张表

2.数据标准

2.1.输入数据格式

实际环境下，各数据会以JSON格式组成，发送到Kafka的一个topic里；此次比赛提供了一个Source 文件用来模拟现实的数据生成。

2.2.输出数据格式

经过计算触发了风控规则告警的以JSON格式在控制台输出

例：{“risk_id”:“1”,“risk_msg”:“不同账户绑定了同一张银行卡！user_id：userId1, userId2；card_no：1234567890123”,“b_user_id”:[“12345678”, “987654321”]}

2.3.数据主键及关系

3.表详细

1.order_info

2.bill_info

3.bill_item

4.ord_pay

5.ord_pay_log

6.pay_method

7.pay_bank_card

4.开发工具、语言版本、工具版本

IDE:idea
scala sdk 2.12.11
java jdk 1.8
flink 1.13.0

5.赛题解答

以下解答，只是我个人的一些解答方案，仅供参考。如果你有好的建议，可以在评论区告诉我。
相应的POJO类见6附录

数据分流

将表进行整理后得到如下有用信息

  表               字段order_info       ORDER_ID,B_USER_ID,TOTAL_PAY,ADD_TIMEbill_info        BILL_ID,ORDER_IDbill_item        BILL_ID,ORD_PAY_IDorder_pay        ORD_PAY_ID,PAY_METHOD_IDpay_method       PAY_METHOD_ID,PAY_RELATE_IDpay_bank_card    PAY_BANK_CARD_ID,B_USER_ID,CARD_NO

数据分流，将多信息流从发送流中分离出来。

 		//支付银行卡流(pay_method,pay_bank_card)OutputTag payCardOutputTag = new OutputTag("paycard") {};//订单流OutputTag billInfoOutputTag = new OutputTag("billinfo") {};//订单明细流OutputTag billItemOutputTag = new OutputTag("billitem") {};//订单支付流OutputTag orderPayOutputTag = new OutputTag("orderpay") {};//订单支付日志流OutputTag orderPayLogOutputTag = new OutputTag("orderpaylog") {};//支付主流costStream(orderId,bUserId,totalPay,addTime)SingleOutputStreamOperator costStream = inputStream.process(new ProcessFunction()  {@Overridepublic void processElement(String s, Context context, Collector collector) throws Exception {JSONObject jsonObject1 = JSONObject.parseObject(s);String orderInfo = jsonObject1.getString("order_info");String payMethod = jsonObject1.getString("pay_method");String payBankCard = jsonObject1.getString("pay_bank_card");String billInfo = jsonObject1.getString("bill_info");String billItem = jsonObject1.getString("bill_item");String orderPay = jsonObject1.getString("ord_pay");String orderPayLog = jsonObject1.getString("ord_pay_log");if (null != orderInfo) {JSONObject orderJson = JSONObject.parseObject(orderInfo);Integer status=orderJson.getInteger("ORD_STATUS");//只需要已支付状态的数据if(status==2) {collector.collect(new OrderInfo(orderJson.getString("ORDER_ID"), orderJson.getString("B_USER_ID"), orderJson.getDouble("TOTAL_PAY"), orderJson.getString("ADD_TIME")));}} else if (null != payMethod) {JSONObject payCardJson = JSONObject.parseObject(payMethod);String payMethodId = payCardJson.getString("PAY_METHOD_ID");String payBankCardId = payCardJson.getString("PAY_RELATE_ID");//等于0，则为卡.如果对应的支付方式是卡，那么我们将cardNO加入支付银行卡流中if(payCardJson.getInteger("PAY_TYPE_FLAG")==0){if(checkPayIsCard.containsKey(payBankCardId)) {//设置payMethodIdPayCard payCard = checkPayIsCard.get(payBankCardId);payCard.setPayMethodId(payMethodId);//加入payCard流context.output(payCardOutputTag,payCard);//清除该支付状态checkPayIsCard.remove(payBankCardId);}}else {//清除该支付状态checkPayIsCard.remove(payBankCardId);}}else if(null != payBankCard){//可以改为和payMethodStream的双流join.但是注意时间延迟。多流join时需要注意等待时间等等JSONObject payCard = JSONObject.parseObject(payBankCard);String payBankCardId = payCard.getString("PAY_BANK_CARD_ID");checkPayIsCard.put(payBankCardId, new PayCard("",payCard.getString("B_USER_ID"), payCard.getString("CARD_NO")));}else if(null!=billInfo){JSONObject billJson = JSONObject.parseObject(billInfo);context.output(billInfoOutputTag,new BillInfo(billJson.getString("BILL_ID"),billJson.getString("ORDER_ID")));}else if(null!=billItem){JSONObject billItemJson = JSONObject.parseObject(billItem);context.output(billItemOutputTag,new BillItem(billItemJson.getString("BILL_ID"),billItemJson.getString("ORD_PAY_ID")));}else if(null!=orderPay){JSONObject orderPayJson = JSONObject.parseObject(orderPay);context.output(orderPayOutputTag,new OrderPay(orderPayJson.getString("ORD_PAY_ID"),orderPayJson.getString("PAY_METHOD_ID")));}else if(null!=orderPayLog){JSONObject orderPayLogJson = JSONObject.parseObject(orderPayLog);context.output(orderPayOutputTag,new OrderPay(orderPayLogJson.getString("ORD_PAY_ID"),orderPayLogJson.getString("PAY_METHOD_ID")));}}});//获取支付银行卡流DataStream payCardStream = costStream.getSideOutput(payCardOutputTag);//获取订单流DataStream billInfoStream = costStream.getSideOutput(billInfoOutputTag);//获取订单明细流DataStream billItemStream = costStream.getSideOutput(billItemOutputTag);//获取订单支付流DataStream orderPayStream = costStream.getSideOutput(orderPayOutputTag);//获取订单支付日志流DataStream orderPayLogStream = costStream.getSideOutput(orderPayLogOutputTag);

其中也可以换成filter操作对数据流一个一个的过滤来得到分流。

任务1

单个用户在一天内，付款超过1万元
其中costStream是已经支付状态的数据流

SingleOutputStreamOperator dangerousMessage1Stream = costStream.keyBy("bUserId").process(new KeyedProcessFunction() {private ValueState costState;private ValueState timeState;@Overridepublic void open(Configuration parameters) throws Exception {costState = getRuntimeContext().getState(new ValueStateDescriptor("cost-value", Double.class));timeState = getRuntimeContext().getState(new ValueStateDescriptor< String>("time-value",  String.class));}@Overridepublic void processElement(OrderInfo orderInfo, Context ctx, Collector collector) throws Exception {String buyUser = orderInfo.getbUserId();Double cost = orderInfo.gettotalPay();String now = orderInfo.getaddTime();//只看支付大于0的账单，避免重复报警if (cost!=0.0) {if (costState.value() == null) {costState.update(cost);timeState.update(now);if (cost > costLimit) {collector.collect(new DangerousMessage("1", "0", buyUser,cost,now));}}else {String start = timeState.value();Double lastCost=costState.value();SimpleDateFormat fmt = new SimpleDateFormat("yyyyMMdd");//如果为同一天if (fmt.parse(start).equals(fmt.parse(now))) {Double costSum = cost + lastCost;costState.update(costSum);if (costSum > costLimit) {/*一天内出现多次总和超过1万，应该多次警报，不清空状态(比如在一段时间,在1万以上还持续增加,则可认为非常高风险,我们可能需要打电话等方式提醒用户,所以这里选择不清空状态)*/collector.collect(new DangerousMessage("1", "0", buyUser,costSum,now));}} else {costState.update(cost);timeState.update(now);}}}}@Overridepublic void close() throws Exception {costState.clear();timeState.clear();}});

任务2

在绑卡操作中发现2个不同用户绑定了同一张银行卡
其中payCardStream是pay_method和ord_pay的关联流

SingleOutputStreamOperator dangerousMessage2Stream = payCardStream.keyBy("cardNo").process(new KeyedProcessFunction() {private ListState payCardState;@Overridepublic void open(Configuration parameters) throws Exception {payCardState = getRuntimeContext().getListState(new ListStateDescriptor("pay-value", String.class));}@Overridepublic void processElement(PayCard payCard, Context ctx, Collector collector) throws Exception {String payCardUser = payCard.getbUserId();String cardNo = payCard.getcardNo();if(isEmpty(payCardState.get())){payCardState.add(payCardUser);}else{Iterable users = payCardState.get();boolean flag=false;for (String user:users) {if(payCardUser.equals(user)){flag=true;break;}}if(!flag) {String conflictUsers ="";for (String user:users) {conflictUsers+="\""+user+"\",";}conflictUsers+="\""+payCardUser+"\"";payCardState.add(payCardUser);collector.collect(new DangerousMessage("2", cardNo,conflictUsers,0.0,""));}}}@Overridepublic void close() throws Exception {payCardState.clear();}});

任务3

在付款过程中，发现单个用户在5分钟内使用了10张以上的银行卡付款
使用table sql进行多流join操作

		//使用table sql进行多流joinTable orderInfoTable = tableEnv.fromDataStream(costStream);Table payCardTable = tableEnv.fromDataStream(payCardStream);Table billInfoTable = tableEnv.fromDataStream(billInfoStream);Table billItemTable = tableEnv.fromDataStream(billItemStream);Table orderPayTable = tableEnv.fromDataStream(orderPayStream);Table orderPayLogTable = tableEnv.fromDataStream(orderPayLogStream);tableEnv.createTemporaryView("order_info",orderInfoTable );tableEnv.createTemporaryView("pay_card",payCardTable );tableEnv.createTemporaryView("bill_info",billInfoTable );tableEnv.createTemporaryView("bill_item",billItemTable );tableEnv.createTemporaryView("order_pay",orderPayTable );tableEnv.createTemporaryView("order_pay_log",orderPayLogTable );//多流join,这里需要注意不能再直接使用costStream来做join操作，而是orderinfo的支付过程和已支付的所有数据。最好的做法是将costStream流的筛选工作直接放到任务一中，而非放到最开始的分流操作中。String sql="select order_info.bUserId bUserId,order_info.addTime addTime,pay_card.cardNo cardNo from order_info " +"join bill_info on order_info.orderId=bill_info.orderId " +"join bill_item on bill_info.billId=bill_item.billId " +"join order_pay on bill_item.orderPayId=order_pay.orderPayId " +"join pay_card on  order_pay.payMethodId=pay_card.payMethodId  ";Table resultTable = tableEnv.sqlQuery(sql);DataStream> joinPayCardStream = tableEnv.toRetractStream(resultTable, OverPayCard.class);
//        joinPayCardStream.print("join");SingleOutputStreamOperator dangerousMessage3Stream = joinPayCardStream.map(line -> {return line.f1;}).assignTimestampsAndWatermarks(new AscendingTimestampExtractor() {@Overridepublic long extractAscendingTimestamp(OverPayCard o) {String addTime = o.addTime;Calendar c = Calendar.getInstance();try {c.setTime(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(addTime));} catch (ParseException e) {e.printStackTrace();}return c.getTimeInMillis();}}).keyBy("bUserId")//滑动窗口设置太小报风险的情况会大量重复，所以这里设置1 minute.window(SlidingEventTimeWindows.of(Time.minutes(windowSizeLimit), Time.minutes(windowSlideSizeLimit))).aggregate(new MyaggregateFunction(), new CountBankCardsFunction());

聚合操作类：

 public static class MyaggregateFunction implements AggregateFunction, Set> {@Overridepublic Set createAccumulator() {return new HashSet<>();}@Overridepublic Set add(OverPayCard overPayCard, Set strings) {strings.add(overPayCard.cardNo);return strings;}@Overridepublic Set getResult(Set strings) {return strings;}@Overridepublic Set merge(Set strings, Set acc1) {strings.addAll(acc1);return strings;}}public static class CountBankCardsFunction implements WindowFunction, DangerousMessage, Tuple, TimeWindow> {@Overridepublic void apply(Tuple tuple, TimeWindow window, Iterable> input, Collector out) throws Exception {String bUserId=tuple.getField(0);Integer count=input.iterator().next().size();if(count>cardLimit){//应该改为count>=cardLimitString bankCards ="";Integer l=0;for(String card:input.iterator().next()){bankCards+="\""+card+"\"";l++;if(lbankCards+=',';}}out.collect(new DangerousMessage("3",bankCards,bUserId,0.0,""));}}}

小结

问题一、二的问题不是很大，不过很多地方还是可以继续优化的。问题三,解答存在一些问题，比如这里其实不能直接使用第一问中的costream作为其中的一个join流，因为第一问中要求的是付款1万元，所以需要使用已支付状态的数据。但是第三问中，题目提示“在付款过程中”，这句话表明只要有开始使用卡就要算在其内,而不只是已支付状态。另外这里和主办方工程师沟通，这里还有个问题是10张以上银行卡是包括10张的，所以count>cardLimit需要改为>=。
当上述两个问题改过之后，能够输出一些数据，但是不全，可能会存在多流join等待时间（因为这些流并非同时到达的。这方面我了解比较少，所以欢迎大家给出自己的见解）的相关问题。

附录

package riskAssessment.Bean;public class BillInfo {public  String billId;public  String orderId;public BillInfo() {}public BillInfo(String billId, String orderId) {this.billId = billId;this.orderId = orderId;}public String getBillId() {return billId;}public void setBillId(String billId) {this.billId = billId;}public String getOrderId() {return orderId;}public void setOrderId(String orderId) {this.orderId = orderId;}@Overridepublic String toString() {return "BillInfo{" +"billId='" + billId + '\'' +", orderId='" + orderId + '\'' +'}';}
}

package riskAssessment.Bean;public class BillItem {public String billId;public String orderPayId;public BillItem() {}public BillItem(String billId, String orderPayId) {this.billId = billId;this.orderPayId = orderPayId;}public String getBillId() {return billId;}public void setBillId(String billId) {this.billId = billId;}public String getOrderPayId() {return orderPayId;}public void setOrderPayId(String orderPayId) {this.orderPayId = orderPayId;}@Overridepublic String toString() {return "BillItem{" +"billId='" + billId + '\'' +", orderPayId='" + orderPayId + '\'' +'}';}
}

package riskAssessment.Bean;public class DangerousMessage {public DangerousMessage() {}public String risk_id;public String card_no;public String b_user_id;public Double totoal_pay;public String time;public DangerousMessage(String risk_id, String card_no, String b_user_id, Double totoal_pay, String time) {this.risk_id = risk_id;this.card_no = card_no;this.b_user_id = b_user_id;this.totoal_pay = totoal_pay;this.time = time;}public Double getTotoal_pay() {return totoal_pay;}public void setTotoal_pay(Double totoal_pay) {this.totoal_pay = totoal_pay;}public String getTime() {return time;}public void setTime(String time) {this.time = time;}public String getRisk_id() {return risk_id;}public void setRisk_id(String risk_id) {this.risk_id = risk_id;}public String getCard_no() {return card_no;}public void setCard_no(String card_no) {this.card_no = card_no;}public String getB_user_id() {return b_user_id;}public void setB_user_id(String b_user_id) {this.b_user_id = b_user_id;}@Overridepublic String toString() {if(risk_id=="1"){return "{\"risk_id\":\"" + risk_id + "\"" +",\"risk_msg\":\"单个用户在一天内，付款超过1万元！" +"\""+",\"b_user_id\":\""+ b_user_id +"\""+",\"time\":\"" +time+"\"" +",\"total_pay\":\"" +totoal_pay.toString()+"\"" +"}";}else if(risk_id=="2"){return "{\"risk_id\":\"" + risk_id + "\"" +",\"risk_msg\":\"不同账户绑定了同一张银行卡！"  + "\"" +",\"card_no\":\"" + card_no + '\"' +",\"b_user_id\":[" + b_user_id + "]}";}else{return "{\"risk_id\":\"" + risk_id + "\"" +",\"risk_msg\":\"单个用户在5分钟内使用了10张以上的银行卡付款！"  + "\"" +",\"b_user_id\":\"" + b_user_id+'\"'+",\"card_no\":[" + card_no  + "]}";}}
}

package riskAssessment.Bean;public class OrderInfo{public String orderId;public String bUserId;public Double totalPay;public String addTime;public OrderInfo() {}public OrderInfo(String orderId, String bUserId, Double totalPay, String addTime) {this.orderId = orderId;this.bUserId = bUserId;this.totalPay = totalPay;this.addTime = addTime;}public String getorderId() {return orderId;}public void setorderId(String orderId) {this.orderId = orderId;}public String getbUserId() {return bUserId;}public void setbUserId(String bUserId) {this.bUserId = bUserId;}public Double gettotalPay() {return totalPay;}public void settotalPay(Double totalPay) {this.totalPay = totalPay;}public String getaddTime() {return addTime;}public void setaddTime(String addTime) {this.addTime = addTime;}@Overridepublic String toString() {return "OrderInfo{" +"orderId='" + orderId + '\'' +", bUserId='" + bUserId + '\'' +", totalPay=" + totalPay +", addTime=" + addTime +'}';}
}

package riskAssessment.Bean;public class OrderPay {public String orderPayId;public String payMethodId;public OrderPay() {}public OrderPay(String orderPayId, String payMethodId) {this.orderPayId = orderPayId;this.payMethodId = payMethodId;}public String getOrderPayId() {return orderPayId;}public void setOrderPayId(String orderPayId) {this.orderPayId = orderPayId;}public String getPayMethodId() {return payMethodId;}public void setPayMethodId(String payMethodId) {this.payMethodId = payMethodId;}@Overridepublic String toString() {return "OrderPay{" +"orderPayId='" + orderPayId + '\'' +", payMethodId='" + payMethodId + '\'' +'}';}
}

package riskAssessment.Bean;public class OrderPayLog {public String orderPayId;public String payMethodId;public OrderPayLog(String orderPayId, String payMethodId) {this.orderPayId = orderPayId;this.payMethodId = payMethodId;}public OrderPayLog() {}public String getOrderPayId() {return orderPayId;}public void setOrderPayId(String orderPayId) {this.orderPayId = orderPayId;}public String getPayMethodId() {return payMethodId;}public void setPayMethodId(String payMethodId) {this.payMethodId = payMethodId;}@Overridepublic String toString() {return "OrderPayLog{" +"orderPayId='" + orderPayId + '\'' +", payMethodId='" + payMethodId + '\'' +'}';}
}

package riskAssessment.Bean;public class OverPayCard {public String bUserId;public String addTime;public String cardNo;public OverPayCard() {}public OverPayCard(String bUserId, String addTime, String cardNo) {this.bUserId = bUserId;this.addTime = addTime;this.cardNo = cardNo;}public String getbUserId() {return bUserId;}public void setbUserId(String bUserId) {this.bUserId = bUserId;}public String getAddTime() {return addTime;}public void setAddTime(String addTime) {this.addTime = addTime;}public String getCardNo() {return cardNo;}public void setCardNo(String cardNo) {this.cardNo = cardNo;}@Overridepublic String toString() {return "OverPayCard{" +"bUserId='" + bUserId + '\'' +", addTime=" + addTime +", cardNo='" + cardNo + '\'' +'}';}
}

package riskAssessment.Bean;public class PayCard {public String payMethodId;public String bUserId;public String cardNo;public PayCard() {}public String getPayMethodId() {return payMethodId;}public void setPayMethodId(String payMethodId) {this.payMethodId = payMethodId;}public String getCardNo() {return cardNo;}public void setCardNo(String cardNo) {this.cardNo = cardNo;}public String getbUserId() {return bUserId;}public void setbUserId(String bUserId) {this.bUserId = bUserId;}public String getcardNo() {return cardNo;}public void setcardNo(String cardNo) {this.cardNo = cardNo;}public PayCard(String payMethodId, String bUserId, String cardNo) {this.payMethodId = payMethodId;this.bUserId = bUserId;this.cardNo = cardNo;}@Overridepublic String toString() {return "PayCard{" +"payMethodId='" + payMethodId + '\'' +", bUserId='" + bUserId + '\'' +", cardNo='" + cardNo + '\'' +'}';}
}

大家有任何问题都可以在评论区留言，我一定尽我所能解答。