python - 使用 pyspark 缓存简单 RDD 时出错，而没有缓存代码则工作正常(How to make a class picklable in a Notebook)

Question

我有以下简单的代码，导致有关缓存的错误:

trips_in = sc.textFile("trip_data.csv")
trips = trips_in.map(lambda l: l.split(",")).map(lambda x: parseTrip(x)).cache()

trips.count()

函数parseTrip()获取字符串列表并创建并返回一个Trip类:

class Trip:
  def __init__(self, id, duration):
    self.id = id
    self.duration = duration

我在执行 count() 操作后立即收到错误。但是，如果我删除第二行末尾的 cache() ，一切都会正常工作。根据错误，问题是 Trip 类无法被 pickle:

PicklingError: Can't pickle __main__.Trip: attribute lookup __main__.Trip failed

那么我怎样才能让它变得可挑选(如果它是一个实际的单词)？请注意，我使用的是 Databricks 笔记本，因此我无法为类定义创建单独的 .py 以使其可挑选。

Answer 1

环境不会影响答案 - 如果您想使用自定义类，它必须可以在集群中的每个节点上导入。

• 对于单个模块，您可以轻松使用 SparkContext.addPyFile 以及 GitHub Gist 的 URL (或其他支持的格式:“file in HDFS (or other Hadoop-supported filesystems), or anHTTP, HTTPS or FTP URI ”)

  • 创建要点。
  • 点击原始链接并复制网址。

  • 在笔记本中调用:

    sc.addPyFile(raw_gist_url)
    

• 对于复杂的依赖关系，您可以分发 Egg 文件。

  • 创建Python package using setuptools .

    目录结构:

    .
    ├── setup.py
    └── trip
        └── __init__.py
    

    示例设置文件:

    #!/usr/bin/env python
    
    from setuptools import setup
    
    setup(name='trip',
          version='0.0.1',
          description='Trip',
          author='Jane Doe',
          author_email='jane@example.com',
          url='https://example.com',
          packages=['trip'],)
    

  • 创建egg文件:

    python setup.py bdist_egg
    

    这将创建包含 trip-0.0.1-pyX.Y.egg 文件的 dist 目录

  • 转到 Databricks 仪表板 -> 新建 -> 库并从 dist 目录上传 Egg 文件:

    

  • 将库附加到您要使用的集群。

• 最后，如果您想要的只是记录类型，则可以使用 namedtuple，无需任何其他步骤:

  from collections import namedtuple
  
  Trip = namedtuple('Trip', ['id', 'duration'])