Блог

Я хочу, чтобы QuantileDiscretizer столбцы фрейма данных были в pyspark. Но есть около 4,000 столбцов для преобразования. Итак, я хотел бы использовать multiprocessing метод следующим образом.

 import multiprocessing as mp
import multiprocessing.pool
from pyspark.ml.feature import QuantileDiscretizer

def transform_col(train,col,numBuckets=5):
    '''
    return: df
    '''
    discretizer = QuantileDiscretizer(numBuckets=numBuckets, inputCol=col, outputCol=col   "_bin")
    discretizer = discretizer.fit(train)
    train = discretizer.transform(train)
    return train
# just an example.
train_df = spark.createDataFrame([[1,2],[3,4],[3,5],[8,8],[3,9],[8,1],[7,1]],["a","b"])
pool = mp.Pool(processes=mp.cpu_count() - 1)
# arguments
process_col = ["a","b"]
args = zip([train_df.select(col) for col in process_col],
           [col for col in process_col]
           )
res = dict(zip([col for col in process_col], pool.starmap(transform_col, args)))
for col in res.keys():
    train_df = train_df.withColumn("process_{}".format(col), res[col].select(col)).drop(col)
pool.close()
pool.join()
  

Но я сталкиваюсь со следующей ошибкой.
Ошибка:

 Py4JError: An error occurred while calling o385.__getstate__. Trace:
py4j.Py4JException: Method __getstate__([]) does not exist
    at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:318)
    at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:326)
    at py4j.Gateway.invoke(Gateway.java:274)
    at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
    at py4j.commands.CallCommand.execute(CallCommand.java:79)
    at py4j.GatewayConnection.run(GatewayConnection.java:238)
    at java.lang.Thread.run(Thread.java:748)
  

Итак, есть ли какой-либо метод для решения этой проблемы?