你可以考虑创建一个管道，用与你关心的名字相关的东西作为键（例如eid），这样可以确保所有流入的事件共享这个标识符，并通过同一个有状态的进程函数（例如KeyedProcessFunction）进行路由。
使用有状态函数可以存储共享同一个eid（即KeyedState）的所有对象的重要状态信息，并且可以执行一些基本的扩充，如下所示（一个简单的示例使用了一些通用的JsonObject，但您可以用自己的类替换它）：

// Function that accepts incoming keyed (strings) events 
// (JsonObject) and outputs enriched events (JsonObject)
class EventEnrichmentFunction: KeyedProcessFunction<String, JsonObject, JsonObject>() {

    // Define the state (mapping properties as key-value pairs of strings)
    private lateinit var state: MapState<String, String>

    override fun open(parameters: Configuration) {
        // Initialize the state (just a map of key-value string pairs)
        state = runtimeContext.getMapState(
            MapStateDescriptor("events", TypeInformation.of(String::class.java), TypeInformation.of(String::class.java))
        )
    }

    // Function that accepts an incoming event (JsonObject), 
    // enriches against state, and outputs events (JsonObject)
    override fun processElement(event: JsonObject, ctx: Context, collector: Collector<JsonObject>) {
        if (state.isEmpty) {
            // This is the first time we've seen an event 
            //with this id, so we need to add properties to keep
            // track of what we've seen
            state.put("name", event.name)
        }

        // Check for properties that you want to try and enrich
        if (!event.has("name") && state.containsKey("name")) {
            event.addProperty("name", state.get("name"))
        }

        // Return the enriched event
        collector.collect(event)
    }
}

基本上，任何键控的有状态函数都允许您基于特定键（在本例中为eid）保留信息，以便共享同一个键的未来事件可以访问它前面的那些键的状态，并随着时间的推移继续构建未来的匹配事件：

if (state.isEmpty) {
    // This is the first time we've seen an event with this id, 
    // so we need to add properties to keep track of what we've seen
    state.put("name", event.name)
    state.put("another_property", event.another_property)
    state.put("yet_another_one", event.yet_another_one);

}

// TODO: Add checks and conditionally enrich events as needed from state

当定义你的工作本身时，它可能看起来像这样：

events
   .keyBy { e -> e.get("eid") }
   .process(EventEnrichmentFunction())
   .sinkTo(...)