python - 什么样的对象可以作为 Spark RDD 中的元素？

Question

可以传递给 SparkContext.parallelize 来创建 RDD 的元素有哪些限制？更具体地说，如果我在 Python 中创建自定义类，我需要实现哪些方法来确保它在 RDD 中正常工作？我假设它需要实现 __eq__ 和 __hash__ 并且可以选择。还有什么？相关文档的链接将不胜感激。我在任何地方都找不到这个。

Answer 1

严格来说，唯一的硬性要求是类是可序列化的(可挑选的)，尽管对于生命周期仅限于单个任务的对象来说没有必要(既不打乱也不收集/并行化)。

仅当类直接用作随机播放键(作为 byKey 操作中的键)时，才需要一致的 __hash__ 和 __eq__ )或间接(例如 distinct 或 cache)。

此外，类定义必须可在每个工作节点上导入，因此模块必须已存在于 PYTHONPATH 上，或与 pyFiles 一起分发。如果类依赖于 native 依赖项，​​则这些依赖项也必须存在于每个工作节点上。

最后，排序类型必须使用标准 Python 语义进行排序。

总结一下:

• 除了可导入之外没有特殊要求:

  class Foo:
     ...
  
  # objects are used locally inside a single task
  rdd.map(lambda i: Foo(i)).map(lambda foo: foo.get(i))
  

• 必须可序列化:

  # Has to be pickled to be distributed 
  sc.parallelize([Foo(1), Foo(2)])
  
  # Has to be pickled to be persisted
  sc.range(10).map(lambda i: Foo(i)).cache()
  
  # Has to be pickled to be fetched to the driver
  sc.range(10).map(lambda i: Foo(i)).collect()  # take, first, etc.
  

• 必须可哈希:

  # Explicitly used as a shuffle key
  sc.range(10).map(lambda i: (Foo(i), 1)).reduceByKey(add)  # *byKey
  
  # Implicitly used as a shuffle kye
  sc.range(10).map(lambda i: Foo(i)).distinct  # subtract, etc.
  

此外，通过闭包传递的所有变量都必须是可序列化的。