案例來源： https://github.com/apache/flink-training/blob/release-1.14/README_zh.md
案例背景
出租車車程(taxi ride)事件結構
1.每次車程都由兩個事件表示：行程開始(trip start)和行程結束(trip end)，
2.每個事件都由十一個欄位組成：
rideId : Long // 每次車程的唯一id
taxiId : Long // 每一輛出租車的唯一id
driverId : Long // 每一位司機的唯一id
isStart : Boolean // 行程開始事件為 TRUE， 行程結束事件為 FALSE
eventTime : Long // 事件的時間戳
startLon : Float // 車程開始位置的經度
startLat : Float // 車程開始位置的維度
endLon : Float // 車程結束位置的經度
endLat : Float // 車程結束位置的維度
passengerCnt : Short // 乘車人數
出租車車費(taxi fare)事件結構
rideId : Long // 每次車程的唯一id
taxiId : Long // 每一輛出租車的唯一id
driverId : Long // 每一位司機的唯一id
startTime : Long // 車程開始時間
paymentType : String // 現金(CASH)或刷卡(CARD)
tip : Float // 小費
tolls : Float // 過路費
totalFare : Float // 總計車費
案例目標
1.將每次車程的 TaxiRide 和 TaxiFare 記錄連接在一起
2.對于每個不同的 rideId，恰好有三個事件：
TaxiRide START 事件
TaxiRide END 事件
一個 TaxiFare 事件（其時間戳恰好與開始時間匹配）
最終的結果應該是 DataStream<RideAndFare>，每個不同的 rideId 都產生一個 RideAndFare 記錄， 每個 RideAndFare 都應該將某個 rideId 的 TaxiRide START 事件與其匹配的 TaxiFare 配對，
案例流程
核心代碼
- connect 可以將兩個流連接成一個ConnectedStreams, 而且不要求兩個流的資料型別一致
// 從車程事件中過濾中車程開始時間，并按車程標識 rideId 分組
KeyedStream<TaxiRide, Long> rideStream = env.fromSource(rideSource, WatermarkStrategy.noWatermarks(), "ride source")
.filter(ride -> ride.getStart()).keyBy(TaxiRide::getRideId);
// 付車費事件按行程標識 rideId 分組
KeyedStream<TaxiFare, Long> fareStream = env.fromSource(fareSource, WatermarkStrategy.noWatermarks(), "fare source")
.keyBy(TaxiFare::getRideId);
rideStream.connect(fareStream).flatMap(new EnrichmentFunction())
.uid("enrichment") // uid for this operator's state
.name("enrichment") // name for this operator in the web UI
.addSink(new PrintSinkFunction<>());
- 使用ValueState保存事件狀態
public class EnrichmentFunction extends RichCoFlatMapFunction<TaxiRide, TaxiFare, RideAndFare> {
private ValueState<TaxiRide> taxiRideValueState;
private ValueState<TaxiFare> taxiFareValueState;
@Override
public void open(Configuration parameters) throws Exception {
ValueStateDescriptor<TaxiRide> taxiRideDescriptor = new ValueStateDescriptor<TaxiRide>("save-ride", TaxiRide.class);
ValueStateDescriptor<TaxiFare> taxiFareDescriptor = new ValueStateDescriptor<TaxiFare>("save-fare", TaxiFare.class);
taxiRideValueState = getRuntimeContext().getState(taxiRideDescriptor);
taxiFareValueState = getRuntimeContext().getState(taxiFareDescriptor);
}
/**
* 當車程事件到來，檢查車費的taxiFareValueState是否保存有對應行程付費記錄
* 如果有，則匹配輸出，清空狀態
* 如果沒有，則將車程事件保存起來
*/
@Override
public void flatMap1(TaxiRide taxiRide, Collector<RideAndFare> collector) throws Exception {
TaxiFare taxiFare = taxiFareValueState.value();
if (Objects.isNull(taxiFare)) {
taxiRideValueState.update(taxiRide);
} else {
taxiFareValueState.clear();
RideAndFare rideAndFare = new RideAndFare();
rideAndFare.setRide(taxiRide);
rideAndFare.setFare(taxiFare);
collector.collect(rideAndFare);
}
}
/**
* 當付費事件到來，檢查車程的taxiRideValueState是否保存有對應行程車程記錄
* 如果有，則匹配輸出，清空狀態
* 如果沒有，則將付費事件保存起來
*/
@Override
public void flatMap2(TaxiFare taxiFare, Collector<RideAndFare> collector) throws Exception {
TaxiRide taxiRide = taxiRideValueState.value();
if (Objects.isNull(taxiRide)) {
taxiFareValueState.update(taxiFare);
} else {
taxiRideValueState.clear();
RideAndFare rideAndFare = new RideAndFare();
rideAndFare.setRide(taxiRide);
rideAndFare.setFare(taxiFare);
collector.collect(rideAndFare);
}
}
}
- 車程事件流和付費事件流來自Kafka
// 定義出租車-車程資料源
KafkaSource<TaxiRide> rideSource = KafkaSource.<TaxiRide>builder()
.setBootstrapServers("192.168.0.192:9092")
.setTopics("TOPIC_RIDE")
.setGroupId("TEST_GROUP")
.setClientIdPrefix("ride") // 避免kafka clientId重復
.setStartingOffsets(OffsetsInitializer.latest())
.setValueOnlyDeserializer(new TaxiRideDeserialization())
.build();
// 定義出租車-車費資料源
KafkaSource<TaxiFare> fareSource = KafkaSource.<TaxiFare>builder()
.setBootstrapServers("192.168.0.192:9092")
.setTopics("TOPIC_FARE")
.setGroupId("TEST_GROUP")
.setClientIdPrefix("fare") // 避免kafka clientId重復
.setStartingOffsets(OffsetsInitializer.latest())
.setValueOnlyDeserializer(new TaxiFareDeserialization())
.build();
事件格式:
1.車程事件: {"rideId":10086, "taxiId":1, "driverId":2, "isStart":true, "eventTime":1656571391726, "startLon":113.273031, "startLat":23.147103, "endLon":113.268245, "endLat":23.14445, "passengerCnt":1}
2.付費事件: {"rideId":10086, "taxiId":1, "driverId":2, "startTime":1656571391726, "paymentType":"CASH", "tip":0.00, "tolls":10.00, "totalFare":110.00}
完整代碼
https://github.com/Mr-LuXiaoHua/study-flink
程式入口: com.example.datastream.rideandfare.RideAndFareJob
轉載請註明出處，本文鏈接：https://www.uj5u.com/shujuku/499006.html
標籤：大數據