“Pizza Price”流是典型的浓缩数据，因此可以是广播流，您将其连接到“Pizza Order”流并按照The Broadcast State Pattern使用。
或者，您可以展平Pizza Order记录，这样一个记录变成N个，每个记录都有一个披萨，然后按商店和披萨键。然后，您可以通过相同的两个字段为Pizza Price流设置关键字，并连接流。丰富之后，您必须按订单ID键，然后重新创建未展平的记录。
在任何一种情况下，你都必须处理基于时间的加入，你的Pizza Price流存储在（比如）MapState<Long, Float>中，你可以在条目上迭代找到最好的一个（time <=订购时间）。
或者使用Table API来帮助您完成上述操作。请参见临时联接。

正如kkrugler所推荐的，对我来说立即奏效的解决方案是应用广播模式。解决方案如下：

解决方案

主管道

// set up streaming execution environment
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(4);

//DataStreamSource
// without watermarks because state is irrelevant
DataStream<PizzaOrder> pizzaOrderStream = env.fromSource(
                this.pizzaOrderSource,
                WatermarkStrategy.noWatermarks(),
                "Kafka Pizza Order Topic"
);

//DataStreamSource
// without watermarks because state is irrelevant
DataStream<PizzaPrice> pizzaPriceStream = env.fromSource(
                this.pizzaPriceSource,
                WatermarkStrategy.noWatermarks(),
                "Kafka Pizza Price Topic"
);

// broadcast the pizza prices and create the broadcast state
BroadcastStream<PizzaPrice> pizzaPriceBroadcastStream = pizzaPriceStream
                .broadcast(pizzaPriceStateDescriptor);

// connect stream (key or non-keyed possible) - call connect on the non-broadcast stream
DataStream<EnrichedPizzaOrder> enrichedPizzaOrderStream = pizzaOrderStream
                .connect(pizzaPriceBroadcastStream)
                .process(new BroadcastPizzaOrderEnrichmentFunction());

enrichedPizzaOrderStream.sinkTo(sink);

// execute the transformation pipeline
return env.execute("Pizza Order Enriching Example");

字符串

充实函数

public static class BroadcastPizzaOrderEnrichmentFunction extends BroadcastProcessFunction<PizzaOrder, PizzaPrice,
            EnrichedPizzaOrder> {
        @Override
        // context -> read only!
        public void processElement(PizzaOrder pizzaOrder, ReadOnlyContext ctx, Collector<EnrichedPizzaOrder> out) throws Exception {
            ReadOnlyBroadcastState<Tuple2<String, String>, PizzaPrice> state = ctx.getBroadcastState(pizzaPriceStateDescriptor);
            List<EnrichedPizza> enrichedPizzas = new ArrayList<>();
            for (Pizza pizza : pizzaOrder.getPizzas()) {
                Tuple2<String, String> stateKey = new Tuple2<>(pizzaOrder.getShop(), pizza.getPizzaName());
                double pizzaPrice = -1; // filler
                if (state.contains(stateKey)) {
                    pizzaPrice = state.get(stateKey).getPrice();
                }
                EnrichedPizza newPizza = new EnrichedPizza(pizza, pizzaPrice);
                enrichedPizzas.add(newPizza);
            }
            EnrichedPizzaOrder enrichedPizzaOrder = new EnrichedPizzaOrder(pizzaOrder, enrichedPizzas);
            out.collect(enrichedPizzaOrder);
        }

        @Override
        // context -> read and write! - must have some deterministic behaviour across all parallel instances
        public void processBroadcastElement(PizzaPrice pizzaPrice, Context ctx, Collector<EnrichedPizzaOrder> out) throws Exception {
            BroadcastState<Tuple2<String, String>, PizzaPrice> state = ctx.getBroadcastState(pizzaPriceStateDescriptor);
            state.put(
                    new Tuple2<>(pizzaPrice.getShop(), pizzaPrice.getName()),
                    pizzaPrice
            );
        }
    }

型

Map描述符

// a map descriptor to store the joined key <shop, pizza> and the associated pizza price object
public static MapStateDescriptor<Tuple2<String, String>, PizzaPrice> pizzaPriceStateDescriptor =
            new MapStateDescriptor<Tuple2<String, String>, PizzaPrice>(
                    "PizzaPriceBroadcastState",
                    TypeInformation.of(new TypeHint<Tuple2<String, String>>() {}),
                    TypeInformation.of(new TypeHint<PizzaPrice>() {}));

型
希望这对你有帮助！