Spark mit Cython

Question

Ich wollte vor kurzem Cython mit Spark verwenden, für die ich folgte the following reference.

Ich schrieb die folgenden Programme wie erwähnt, aber ich bin immer ein:

TypeError: 
fib_mapper_cython() takes exactly 1 argument (0 given) 

spark-tools.py

def spark_cython(module, method): 
    def wrapped(*args, **kwargs): 
     global cython_function_ 
     try: 
      return cython_function_(*args, **kwargs) 
     except: 
      import pyximport 
      pyximport.install() 
      cython_function_ = getattr(__import__(module), method) 
     return cython_function_(*args, **kwargs) 
    return wrapped() 

fib.pyx

def fib_mapper_cython(n): 
    ''' 
    Return the first fibonnaci number > n. 
    ''' 
    cdef int a = 0 
    cdef int b = 0 
    cdef int j = int(n) 
    while b<j: 
     a, b = b, a+b 
    return b, 1 

main.py

from spark_tools import spark_cython 
import pyximport 
import os 
from pyspark import SparkContext 
from pyspark import SparkConf 
pyximport.install() 


os.environ["SPARK_HOME"] = "/home/spark-1.6.0" 
conf = (SparkConf().setMaster('local').setAppName('Fibo')) 

sc = SparkContext() 
sc.addPyFile('file:///home/Cythonize/fib.pyx') 
sc.addPyFile('file:///home/Cythonize/spark_tools.py') 
lines = sc.textFile('file:///home/Cythonize/nums.txt') 

mapper = spark_cython('fib', 'fib_mapper_cython') 
fib_frequency = lines.map(mapper).reduceByKey(lambda a, b: a+b).collect() 
print fib_frequency 

Ich bekomme eine TypeError immer wenn ich das Programm starte. Irgendwelche Ideen?

Answer 1

Dies ist kein Cython noch ein PySpark Problem, Sie haben leider einen zusätzlichen Funktionsaufruf während der Definition von spark_cython hinzugefügt. Insbesondere ist die Funktion, die den Aufruf der cython_function hüllt wird ohne Argumente auf Rückkehr genannt:

return wrapped() # call made, no args supplied. 

Als Ergebnis werden Sie nicht die verpackte Funktion zurück, wenn Sie diesen Anruf auszuführen. Was Sie tun, ist wrapped ohne *args oder **kwargs anrufen. wrapped ruft dann fib_mapper_cython ohne Argumente (seit *args, **kwargs sind nicht im Lieferumfang enthalten) daher die TypeError.

Sie sollten stattdessen:

return wrapped 

und dieses Problem nicht mehr vorhanden sein sollte.