flink真实案例-读取kafka下沉redis-中间逻辑处理带原始数据.zip

package com.bitz.coin.benifit.handle; import com.alibaba.fastjson.JSONObject; import com.bitz.coin.benifit.utils.JsonUtil; import org.apache.flink.api.common.functions.FlatMapFunction; import org.apache.flink.api.common.functions.ReduceFunction; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.util.Collector; import java.math.BigDecimal; import java.util.ArrayList; /** * 数据的处理 */ public class Handle { private static DataStream<String> dataStream; private Handle(DataStream<String> dataStream1) { dataStream = dataStream1; } private static Handle handle; public static Handle getInstance(DataStream<String> dataStream) { if (handle == null) { handle = new Handle(dataStream); } return handle; } public DataStream<Num> getmap() { return dataStream .flatMap(new FlatMapFunction<String, Num>() { @Override public void flatMap(String in, Collector<Num> out) throws Exception { if (in.contains("order_coin")) { ArrayList<Num> arrayList = parseJson(in); for (Num numline : arrayList) { out.collect(numline); } } } }) //此处key为Num中的字段key .keyBy("key") .reduce(new ReduceFunction<Num>() { @Override //value1上一次累加的num, value2这一次的num public Num reduce(Num value1, Num value2) throws Exception { return new Num(value1.getKey(), value1.getNum().add(value2.getNum())); } }); } /** * 此处为解析程序 * */ private static ArrayList parseJson(String in) { ArrayList<Num> arrayList = new ArrayList(); JSONObject jsonObject1 = JsonUtil.getData(in); //买方 Object buy_uid1 = jsonObject1.get("buy_uid"); Object coin_from1 = jsonObject1.get("coin_from"); Object number3 = jsonObject1.get("number"); Object sale_uid1 = jsonObject1.get("sale_uid"); Object coin_to1 = jsonObject1.get("coin_to"); BigDecimal number1 = new BigDecimal(0); if (null != buy_uid1 && null != coin_from1 && null != number3) { String buy_uid = buy_uid1.toString(); String coin_from = coin_from1.toString(); number1 = new BigDecimal(number3.toString()); String buy = buy_uid + "_" + coin_from; Num buyNum = new Num(buy, number1); arrayList.add(buyNum); } //卖方 if (null != sale_uid1 && null != coin_to1) { String sale_uid = sale_uid1.toString(); String coin_to = coin_to1.toString(); BigDecimal price = new BigDecimal(jsonObject1.get("price").toString()); BigDecimal number2 = number1.multiply(price); Num saleNum = new Num(sale_uid + "_" + coin_to, number2); arrayList.add(saleNum); } return arrayList; } }