import rpy2.robjects as robjects

dffunc = sc.parallelize([(0,robjects.r.rnorm),(1,robjects.r.runif)])
dffunc.collect() 

Ausgänge

[(0, <rpy2.rinterface.SexpClosure - Python:0x7f2ecfc28618 / R:0x26abd18>), (1, <rpy2.rinterface.SexpClosure - Python:0x7f2ecfc283d8 / R:0x26aad28>)]

Während die partitionierte Version zu einem Fehler führt:

dffuncpart = dffunc.partitionBy(2)
dffuncpart.collect()

RuntimeError: ('R cannot evaluate code before being initialized.', <built-in function unserialize>

Es scheint, als wäre dieser Fehler, dassR wurde nicht auf eine der Partitionen geladen, was impliziert, dass der erste Importschritt nicht ausgeführt wurde. Gibt es sowieso darum herum?

EDIT 1 Dieses zweite Beispiel lässt mich denken, dass es einen Fehler im Timing von pyspark oder rpy2 gibt.

dffunc = sc.parallelize([(0,robjects.r.rnorm),     (1,robjects.r.runif)]).partitionBy(2)
def loadmodel(model):
    import rpy2.robjects as robjects
    return model[1](2)
dffunc.map(loadmodel).collect()

Erzeugt den gleichen Fehler, den R vor der Initialisierung nicht auswerten kann.

dffuncpickle = sc.parallelize([(0,pickle.dumps(robjects.r.rnorm)),(1,pickle.dumps(robjects.r.runif))]).partitionBy(2)
def loadmodelpickle(model):
    import rpy2.robjects as robjects
    import pickle
    return pickle.loads(model[1])(2)
dffuncpickle.map(loadmodelpickle).collect()

Works genau wie erwartet.