apache-spark - 在 Spark 中的 map 函数内运行 ML 算法

Question

所以这几天我一直在尝试在 Spark 的映射函数中运行机器学习算法。我发布了更具体的question但引用 Spark 的 ML 算法会出现以下错误:

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

显然我无法在 apply_classifier 函数内引用 SparkContext 。我的代码与我提出的上一个问题中建议的类似，但仍然没有找到我正在寻找的解决方案:

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

is it possible to do what I am trying to do?

事实并非如此。 Apache Spark 不支持任何形式的嵌套，分布式操作只能由驱动程序初始化。这包括访问分布式数据结构，例如 Spark DataFrame。

What are the alternatives?

这取决于许多因素，例如数据大小、可用资源量以及算法选择。一般来说，您有三个选择:

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

  一般来说，当数据相对较小时，这种方法特别有用。它的优点是多个工作之间不存在竞争。

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

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

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

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