将 RDD 转为广播字典进行查找答案

【问题标题】：Turn RDD into Broadcast Dictionary for lookup将 RDD 转为广播字典进行查找
【发布时间】：2015-10-16 05:18:21
【问题描述】：

到目前为止我所拥有的是：

lookup = sc.textFile("/user/myuser/lookup.asv")
lookup.map(lambda r: r.split(chr(1)) )

现在我的 RDD 看起来像

[
    [filename1, category1],
    [filename2, category2],
    ...
    [filenamen, categoryn]
]

如何将该 RDD 转换为广播字典，例如：

{filename1: category1, filename2: category2, ...}

这是我尝试过但不起作用的方法：

>>> broadcastVar = sc.broadcast({})
>>> data = sc.parallelize([[1,1], [2,2], [3,3], [4,4]])
>>> def myfunc(x):
...     broadcastVar[str(x[0])] = x[1]
... 
>>> result = data.map(myfunc)
>>> broadcastVar
<pyspark.broadcast.Broadcast object at 0x7f776555e710>
>>> broadcastVar.value
{}
>>> result.collect()
...
ERROR: TypeError: 'Broadcast' object does not support item assignment
...
>>> broadcastVar.value
{}

有关我为什么要构建这个巨大的查找变量的更多信息，请阅读以下内容：

这是one 的后续问题。

我有两张桌子

table1：一个非常宽（25K 列和 150K 行）的表，其中每列包含像素信息，第一列是输入图像文件的文件名。

table2：一个 TSV（制表符分隔文件）文件，有 300 万行，每行包含图像文件名和图像的产品类别。

在 SQL 中，我需要对文件名上的这两个表进行内部连接，以便我可以标记图像数据以供以后进行机器学习。

在任何类型的 SQL 中执行此操作都是不现实的，因为您必须为 table1 创建一个具有 25K 列的表，创建表的语法会很长。

然后我正在考虑使用 table2 创建一个查找变量，并可能使其成为一个广播变量，其中键是文件名，值是产品类别。

【问题讨论】：

标签： python apache-spark

【解决方案1】：

广播变量在工作人员上是只读的。 Spark 提供了只写的累加器，但这些累加器用于计数器之类的东西。在这里你可以简单地收集和创建一个 Python 字典：

lookup_bd = sc.broadcast({
  k: v for (k, v) in lookup.map(lambda r: r.split(chr(1))).collect()
})

在任何类型的 SQL 中执行此操作都是不现实的，因为您必须为 table1 创建一个具有 25K 列的表，创建表的语法会很长。

创作应该不是问题。只要您知道名称，您就可以轻松地以编程方式创建这样的表：

from pyspark.sql import Row

colnames = ["x{0}".format(i) for i in range(25000)] # Replace with actual names

df = sc.parallelize([
   row(*[randint(0, 100) for _ in range(25000)]) for x in range(10)
]).toDF()

## len(df.columns)
## 25000

这里还有一个更严重的问题，即使你使用普通的 RDD。非常宽的行通常很难以任何逐行格式处理。

您可以做的一件事是使用稀疏表示，例如 SparseVector 或 SparseMatrix。另一种是使用 RLE 对像素信息进行编码。

【讨论】：