在Spark中的map函数中运行ML算法

Question

所以我现在已经尝试了几天在Spark中的map函数中运行ML算法.我发布了一个更具体的问题,但引用Spark的ML算法给出了以下错误:

AttributeError: Cannot load _jvm from SparkContext. Is SparkContext initialized?

显然我无法SparkContext在apply_classifier函数内部引用.我的代码与我之前提出的问题中的建议类似,但仍未找到我正在寻找的解决方案:

def apply_classifier(clf):
    dt = DecisionTreeClassifier(labelCol="indexedLabel", featuresCol="indexedFeatures", maxDepth=3)
    if clf == 0:
        clf = DecisionTreeClassifier(labelCol="indexedLabel", featuresCol="indexedFeatures", maxDepth=3)
    elif clf == 1:
        clf = RandomForestClassifier(labelCol="indexedLabel", featuresCol="indexedFeatures", numTrees=5)

classifiers = [0, 1]

sc.parallelize(classifiers).map(lambda x: apply_classifier(x)).collect() 

我尝试使用flatMap而不是,map但我得到NoneType object is not iterable.

我还希望将广播的数据集(这是一个DataFrame)作为函数内的参数传递apply_classifier.最后,是否有可能做我想做的事情？有哪些替代方案？

Answer 1

我可以做我想做的事情吗？

它不是.Apache Spark不支持任何形式的嵌套,分布式操作只能由驱动程序初始化.这包括访问分布式数据结构,如Spark DataFrame.

有哪些替代方案？

这取决于许多因素,如数据大小,可用资源量和算法选择.一般来说,您有三种选择:

• 仅使用Spark作为任务管理工具来训练本地的非分布式模型.看起来你已经在某种程度上探索了这条道路.有关此方法的更高级实现,您可以检查spark-sklearn.

  通常,当数据相对较小时,这种方法特别有用.它的优点是多个工作之间没有竞争.

• 使用标准多线程工具从单个上下文提交多个独立作业.你可以使用例如threading或joblib.

  虽然这种方法是可行的,但我不会在实践中推荐它.并非所有Spark组件都是线程安全的,您必须非常小心以避免意外行为.它还使您几乎无法控制资源分配.

• 参数化您的Spark应用程序并使用外部管道管理器(Apache Airflow,Luigi,Toil)提交您的作业.

  虽然这种方法有一些缺点(它需要将数据保存到持久存储),但它也是最通用和最强大的,并且可以对资源分配进行大量控制.