python - PySpark:如何通过多处理并行处理数据帧的多列?
问题描述
我想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)
那么,有没有什么方法可以解决这个问题呢?
解决方案
推荐阅读
- javascript - 选择对象更改选定值
- android - 如何将现有的异步请求与 RxAndroid 或使用 Kotlin 协程结合?
- android-studio - React-native (Signed) release apk 不断在设备中崩溃
- html - 在 Safari 和 iPad Chrome 中不可见的相对定位链接
- go - 如何将字符串转换为 int128 的值?目前我仅限于 int64
- sql-server - 为什么共享锁允许在 SQL Server 中进行更新?
- javascript - 尝试在相同的键名下获取多个值,而不是获取键的最后一个值
- java - mock-maker-inline 使其他模拟不起作用
- ruby - * 运算符在属性数组之前做什么?
- c++ - 如何在 Windows 命令提示符下使用 Dev C++ 编译器编译 C++ 程序?