java - 为不同的事件构建状态链并在 spark 中分配全局 ID

Question

我们正在与 spark 1.6 合作我们正在努力保持类似事件的全局身份。可以有几个具有相同 ID 的事件“组”(在示例中为数字。添加字母只是为了唯一性)。我们知道其中一些事件是相似的，因此我们能够将它们联系起来。我们想保留这样的东西:

Z -> 1, 2, 3
X -> 4

所以将来如果有 id 为 4 的事件发生，我们可以分配 X作为全局身份。

请检查示例以获得更好的说明:

假设我们有一些流数据进入 spark 作业。

1a
1b
2c
2d
2e
3f
3g
3h
4i

由于事件 1 是我们的第一次亮相，我们要分配 1 to Z . 接下来我们知道 1b 和 2c 是相似的。所以我们想保留在某个地方 2->1映射。 2e 和 3f 也是一样，所以我们需要映射 3-2 .所以现在我们有 3 对 1->Z , 2->1 , 3->2 .

我们要创建“历史”路径:Z <- 1 <- 2 <- 3 最后，我们将使用 ID = Z 处理所有事件。 .

1a -> Z
1b -> Z
2c -> Z
2d -> Z
2e -> Z
3f -> Z
3g -> Z
3h -> Z
4i -> X

我们尝试使用 mapwithstate但我们唯一能做的就是2->1和 3->2 .与 mapwithstate我们无法在当前事件的状态中获取“父级”的状态 - 例如。当前事件 3 与父 2 无法获得 2 -> 1也不是1 -> Z .

是否可以为此进行一些全局映射？我们已经尝试过累加器和广播，但看起来不太合适。我们无法用 Z 替换第一个映射的事件 1 和第二个映射的事件 2| .

如果有新事件5会来，它与 3h 类似，例如我们需要分配映射 5->Z再次。

Answer 1

接下来是给定问题的解决方案，使用对“状态”RDD 的可变引用，我们每次都会用新结果更新它。

我们使用transform 通过执行相似性连接，用唯一的全局 id 标记传入的事件流。这是“手动”连接，我们使用两个数据集的乘积并成对比较每个条目。

请注意，这是一个昂贵的过程。有许多部分可以更改，具体取决于预期流的具体特征。例如，我们可以将全局状态 RDD 替换为本地 map 并应用 map-side 连接以获得更快的相似性连接，但这在很大程度上取决于预期的基数唯一 ID 集。

这比我原先预期的要棘手。仅将其作为迈向更强大解决方案的起点。例如，状态 RDD 上的 union 操作需要定期检查点，以避免 DAG 超出控制。 (有很大的改进空间 - 但这超出了提供答案的合理努力。)

我在这里勾勒出解决方案的核心，完整的测试笔记本请参见 UniqueGlobalStateChains.snb

// this mutable reference points to the `states` that we keep across interations    
@transient var states: RDD[(String, (Int, Long))] = sparkContext.emptyRDD

// we assume an incoming Event stream. Here we prepare it for the global id-process 
@transient val eventsById = eventStream.map(event => (event.id, event))
@transient val groupedEvents = eventsById.groupByKey()

// this is the core of the solution. 
// We transform the incoming events into tagged events. 
// As a by-product, the mutable `states` reference will get updated with the latest state mapping. 
// the "chain" of events can be reconstructed ordering the states by timestamp

@transient val taggedEvents = groupedEvents.transform{ (events, currentTime) => 
    val currentTransitions = states.reduceByKey{case (event1, event2) => Seq(event1, event2).maxBy{case (id, ts) => ts}}                        
    val currentMappings = currentTransitions.map{case (globalId, (currentId, maxTx)) => (currentId, globalId)}

    val newEventIds = events.keys // let's extract the ids of the incoming (grouped) events
    val similarityJoinMap = newEventIds.cartesian(currentMappings)
        .collect{case (eventId, (currentId, globalId)) if (isSimilar(currentId)(eventId)) => (eventId, globalId)}
        .collectAsMap
    //val similarityBC = sparkContext.broadcast(similarityJoinMap)                   
    val newGlobalKeys = newEventIds.map(id => (id, similarityJoinMap.getOrElse(id, genGlobalId())))
    newGlobalKeys.cache() //avoid lazy evaluation to generate multiple global ids

    val newTaggedEvents = events.join(newGlobalKeys).flatMap{case (eventId, (events, globalKey)) => 
                                      events.map(event => (event.id,event.payload, globalKey))
                                     }
    val newStates = newGlobalKeys.map{case (eventId, globalKey) => (globalKey, (eventId, currentTime.milliseconds))}
    currentState = newStates              
    states.unpersist(false)                              
    states = newStates.union(states)
    states.cache()                              
    newTaggedEvents
    }

给定这个输入序列:

"1|a,1|b,3|c",  "2|d,2|e,2|f", "3|g,3|h,3|i,4|j", "5|k", "4|f,1|g", "6|h"

我们得到:

具有全局 ID 的标记事件:

---
1|a: gen-4180,1|b: gen-4180,3|c: gen-5819
---
2|d: gen-4180,2|e: gen-4180,2|f: gen-4180
---
3|g: gen-4180,3|h: gen-4180,3|i: gen-4180,4|j: gen-5819
---
5|k: gen-5819
---
1|g: gen-2635,4|f: gen-4180
---
6|h: gen-5819

我们可以重建从全局 id 派生的事件链:

gen-4180: 1<-2<-3<-4
gen-2635: 1
gen-5819: 3<-4<-5<-6

-o-