如何在Spark中的map函数中使用数据帧？

Question

定义:

• sampleDF 是示例数据框,具有用于查找目的的列表记录.
• sampleDS 是一个包含元素列表的RDD.
• mappingFunction是查找sampleDSin 的元素sampleDF并将它们映射到1(如果它们存在),sampleDF如果它们不存在则映射到0.

我有一个映射函数如下:

def mappingFunction(element):
    # The dataframe lookup!
    lookupResult = sampleDF.filter(sampleDF[0] == element).collect()
    if len(lookupResult) > 0:
        print lookupResult
        return 1
    return 0

问题:

访问sampleDF映射函数之外的工作完全正常,但只要我在函数内部使用它,我就会收到以下错误:

py4j.Py4JException: Method __getnewargs__([]) does not exist
    at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:335)
    at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:344)
    at py4j.Gateway.invoke(Gateway.java:252)
    at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:133)
    at py4j.commands.CallCommand.execute(CallCommand.java:79)
    at py4j.GatewayConnection.run(GatewayConnection.java:209)
    at java.lang.Thread.run(Thread.java:744)

我还尝试了什么:

我确实尝试保存一个临时表并使用sqlContextmap函数中的select但仍然无法使其工作.这是我得到的错误:

  File "/usr/lib64/python2.6/pickle.py", line 286, in save
    f(self, obj) # Call unbound method with explicit self
  File "/usr/lib64/python2.6/pickle.py", line 649, in save_dict
    self._batch_setitems(obj.iteritems())
  File "/usr/lib64/python2.6/pickle.py", line 686, in _batch_setitems
    save(v)
  File "/usr/lib64/python2.6/pickle.py", line 331, in save
    self.save_reduce(obj=obj, *rv)
  File "/opt/spark/python/pyspark/cloudpickle.py", line 542, in save_reduce
    save(state)
  File "/usr/lib64/python2.6/pickle.py", line 286, in save
    f(self, obj) # Call unbound method with explicit self
  File "/usr/lib64/python2.6/pickle.py", line 649, in save_dict
    self._batch_setitems(obj.iteritems())
  File "/usr/lib64/python2.6/pickle.py", line 681, in _batch_setitems
    save(v)
  File "/usr/lib64/python2.6/pickle.py", line 306, in save
    rv = reduce(self.proto)
TypeError: 'JavaPackage' object is not callable

我要的是:

我试图通过简单的例子来简化我的问题.任何有关如何在地图功能中使用数据框的帮助都非常受欢迎.

Answer 1

这不可能。Spark 不支持分布式数据结构（RDDs、DataFrames、Datasets）上的嵌套操作。即使它确实执行大量作业也不是一个好主意。鉴于您所展示的代码，您可能希望将 RDD 转换为 aDataFrame并执行join机智

(rdd.map(x => (x, )).toDF(["element"])
  .join(sampleDF, sampleDF[0] == df[0])
  .groupBy("element")
  .agg(count("element") > 0))

顺便说一下，里面的打印map是完全没用的，更不用说它会增加额外的 IO 开销。