如何加载无法在每个 Spark 执行器中腌制的数据？

小能豆

如何加载无法在每个 Spark 执行器中腌制的数据？

我正在使用用Cython 编写的NoAho 库。它的内部 trie 无法被 pickle：如果我在主节点上加载它，我永远无法获得在工作节点中执行的操作的匹配项。

因为我想在每个 Spark 执行器中使用相同的 trie，所以我找到了一种延迟加载 trie 的方法，受到这个spark 上的 spaCy 问题的启发。

global trie

def get_match(text):
    # 1. Load trie if needed
    global trie
    try:
        trie
    except NameError:
        from noaho import NoAho

        trie = NoAho()
        trie.add(key_text='ms windows', payload='Windows 2000')
        trie.add(key_text='ms windows 2000', payload='Windows 2000')
        trie.add(key_text='windows 2k', payload='Windows 2000')
        ...

    # 2. Find an actual match to get they payload back
    return trie.findall_long(text)

虽然这有效，但.add()每个 Spark 作业都会执行所有调用，这大约需要一分钟。由于我不确定“Spark 作业”是否是正确的术语，所以我会更明确：我在 Jupyter 笔记本中使用 Spark，每次运行需要该get_match()功能的单元时，trie 都不会被缓存，并且需要一分钟来加载这些 trie，这占据了运行时间的大部分。

我可以做些什么来确保 trie 被缓存？或者有更好的解决方案来解决我的问题？

阅读 58

2025-02-26

共1个答案

小能豆

您可以尝试使用单例模块来加载和初始化trie。基本上，您需要的只是一个包含以下内容的单独模块：

trie_loader.py

```py
from noaho import NoAho

def load():
trie = NoAho()
trie.add(‘ms windows’, ‘Windows 2000’)
trie.add(‘ms windows 2000’, ‘Windows 2000’)
trie.add(‘windows 2k’, ‘Windows 2000’)
return trie

trie = load()
```

并使用标准 Spark 工具进行分发：

sc.addPyFile("trie_loader.py")
import trie_loader

rdd = sc.parallelize(["ms windows", "Debian GNU/Linux"])
rdd.map(lambda x: (x, trie_loader.trie.find_long(x))).collect()
## [('ms windows', (0, 10, 'Windows 2000')),
##  ('Debian GNU/Linux', (None, None, None))]

这应该在每次启动 Python 进程执行器时加载所需的数据，而不是在访问数据时加载。我不确定它是否有用，但值得一试。

2025-02-26