scala - 基于顶点 ID 创建边引发 scala

Question

我想基本上加入两个 RDD 的顶点和边。使用以下代码创建顶点和边:

val file = sc.textFile("file.gz") //This tab separated file has more than two columns among which only first two columns with source and destination URL are relevant 

val edges= file.flatMap(f => {
  val urls = f.split("\t")
  if (!(urls.length < 2)) 
{ Some(urls(0) +"\t"+ urls(1)) }
else None }).distinct

val vertices = edges.flatMap(f => f.split("\t")).distinct 
val vertices_zip = vertices.zipWithUniqueId

现在我有一个顶点(URL)列表，其中包含使用上述方法生成的 ID，例如:

google.de/2011/10/Extract-host,11
facebook.de/2014/11/photos,28         
community.cloudera.com/t5/,42         
facebook.de/2020/11/photos,91 

我想根据这些 ID 创建边。 Edges RDD 文件以制表符分隔，如下所示:

google.de/2011/10/Extract-host   facebook.de/2014/11/photos   
facebook.de/2014/11/photos       community.cloudera.com/t5/,42
community.cloudera.com/t5/       google.de/2011/10/Extract-host

要求的结果:

11     28
28     42
42     11

我尝试了以下代码

val edges_id = edges.flatMap( line => line.split( "\t" ) ).map( line => ( line,0) ) .join(vert_zip).map(x=>x._2._2)

但是没有得到想要的结果。我正进入(状态

11
28
28
42
42
11

我不确定如何将边与顶点 RDD 连接起来以获得此结果。任何帮助将非常感激。

Answer 1

当您zipWithUniqueId ，然后收集 rdds 作为 map ，然后使用该 map 获取边缘 rdd 中的索引，如下所示

val vertices_zip = vertices.zipWithUniqueId.collectAsMap

val edges_id = edges.map(f => {
  val urls = f.split("\t")
  vertices_zip(urls(0))+"\t"+vertices_zip(urls(1))
})

就这样。我希望答案有帮助

更新

你评论了

I am getting an exception : java.lang.OutOfMemoryError: Java heap space

为此，您可以使用广播，它将调用所需的 rdds 到执行程序内存而不是所有 map

val vertices_zip = sc.broadcast(vertices.zipWithUniqueId.collectAsMap)

val edges_id = edges.map(f => {
  val urls = f.split("\t")
  vertices_zip.value(urls(0))+"\t"+vertices_zip.value(urls(1))
})

加入

你又评论了

Is it possible to change the code I tried above to get the result (the one with the join)?

连接方式需要两个连接，这意味着需要两次洗牌才能获得所需的结果

val vertices_zip = vertices.zipWithUniqueId

val edges_id = edges.map(line => {
  val splitted = line.split("\t")
  (splitted(0), splitted(1))
})
  .join(vertices_zip)
  .map(_._2)
  .join(vertices_zip)
  .map(x => x._2._1+"\t"+x._2._2)