Flink-基于 DataStream API 实现欺诈检测 - ToB企服应用市场:ToB评测及商务社交产业平台

1. 数据源使用Kafka，发送json格式字符串
消息格式: {"accountId":1001, "timestamp":1656490723171, "amount":0.12}
2. 自定义 DeserializationSchema, 直接将kafka的json字符串转成POJO对象

复制代码

public class TransactionDeserialization implements DeserializationSchema<Transaction> {
@Override
public Transaction deserialize(byte[] bytes) throws IOException {
ByteBuffer buffer = ByteBuffer.wrap(bytes);
String message = byteBufferToString(buffer);
if (StringUtils.isBlank(message)) {
return null;
}
Transaction transaction = JsonUtils.fromJson(message, Transaction.class);
return transaction;
}
@Override
public boolean isEndOfStream(Transaction transaction) {
return false;
}
@Override
public TypeInformation<Transaction> getProducedType() {
return TypeInformation.of(Transaction.class);
}
/**
* ByteBuffer 转换 String
* @param buffer
* @return
*/
private String byteBufferToString(ByteBuffer buffer) {
String ret = "";
try{
CharsetDecoder decoder = Charset.forName("UTF-8").newDecoder();
CharBuffer charBuffer = decoder.decode(buffer.asReadOnlyBuffer());;
ret = charBuffer.toString();
}catch (Exception e) {
e.printStackTrace();
}
return ret;
}
}

复制代码

public class FraudDetector extends KeyedProcessFunction<Long, Transaction, Alert> {
/**
* 定义小金额边界
*/
private static final double SMALL_AMOUNT = 1.00;
/**
* 定义大金额边界
*/
private static final double LARGE_AMOUNT = 500.00;
/**
* 1分钟时间
*/
private static final long ONE_MINUTE = 60 * 1000;
/**
* 保存是否有消费小金额的状态
*/
private transient ValueState<Boolean> smallAmountState;
/**
* 定时器状态
*/
private transient ValueState<Long> timerState;
@Override
public void open(Configuration parameters) throws Exception {
// 初始化ValueState
ValueStateDescriptor<Boolean> smallAmountStateDescriptor = new ValueStateDescriptor<Boolean>("small-amount-state", Types.BOOLEAN);
smallAmountState = getRuntimeContext().getState(smallAmountStateDescriptor);
ValueStateDescriptor<Long> timerStateDescriptor = new ValueStateDescriptor<Long>("timer-state", Types.LONG);
timerState = getRuntimeContext().getState(timerStateDescriptor);
}
@Override
public void processElement(Transaction transaction, Context context, Collector<Alert> collector) throws Exception {
if (Objects.isNull(transaction)) {
return;
}
// Get the current state for the current key
Boolean lastTransactionWasSmall = smallAmountState.value();
// Check if the flag is set
if (Objects.nonNull(lastTransactionWasSmall)) {
if (transaction.getAmount() > LARGE_AMOUNT) {
Alert alert = new Alert();
alert.setAccountId(transaction.getAccountId());
alert.setAmount(transaction.getAmount());
collector.collect(alert);
}
clearUp(context);
}
if (transaction.getAmount() < SMALL_AMOUNT) {
// set the flag to true
smallAmountState.update(true);
// 注册定时器，设置一个当前时间一分钟后触发的定时器，同时，将触发时间保存到 timerState 状态中。
long timer = context.timerService().currentProcessingTime() + ONE_MINUTE;
context.timerService().registerProcessingTimeTimer(timer);
timerState.update(timer);
}
}
@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector<Alert> out) throws Exception {
// remove flag after 1 minute
timerState.clear();
smallAmountState.clear();
}
private void clearUp(Context ctx) {
try {
// delete timer
Long timer = timerState.value();
ctx.timerService().deleteProcessingTimeTimer(timer);
timerState.clear();
smallAmountState.clear();
} catch (IOException e) {
e.printStackTrace();
}
}
}

复制代码

public class FraudDetectionJob {
public static void main(String[] args) throws Exception {
// 初始化环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// kafka消息格式: {"accountId":1001, "timestamp":1656490723171, "amount":0.12}
// 定义Kafka数据源
KafkaSource<Transaction> source = KafkaSource.<Transaction>builder()
.setBootstrapServers("192.168.0.192:9092")
.setTopics("TOPIC_FRAUD_DETECTION")
.setGroupId("TEST_GROUP")
.setStartingOffsets(OffsetsInitializer.latest())
.setValueOnlyDeserializer(new TransactionDeserialization())
.build();
// 加载数据源
DataStreamSource<Transaction> fraudDetectionSource
= env.fromSource(source, WatermarkStrategy.noWatermarks(), "FraudDetection-Source");
// 处理数据
SingleOutputStreamOperator<Alert> alertStreamOperator = fraudDetectionSource.keyBy(Transaction::getAccountId)
.process(new FraudDetector())
.name("Fraud-Detector");
// 输出告警结果
alertStreamOperator.addSink(new AlertSink())
.name("Send-Alerts");
env.execute("Fraud Detection");
}
}

复制代码

代码入口: com.example.datastream.frauddetection.FraudDetectionJob

复制代码