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

Question

我们正在努力,spark 1.6我们正在努力保持类似事件的全球身份.可能很少有具有相同ID的事件"组"(在示例中为数字.仅为唯一性添加字母).我们知道其中一些事件是相似的,所以我们可以连接它们.我们希望保留以下内容:

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

因此,如果将来有一些id为4的事件,我们可以指定X为全局身份.

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

假设我们有一些流数据进入火花工作.

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.

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

如果新事件5将来,并且它与3h类似,例如我们需要5->Z再次分配映射.

Answer 1

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

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

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

这比我最初预期的要棘手。仅将其作为更强大解决方案的起点。例如，union对状态 RDD 的操作需要定期检查点，以避免 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-