移植多线程计算密集型作业以激发

Question

我正在编写使用执行器服务来并行化任务的代码（想想机器学习计算一遍又一遍地在小数据集上完成）。 我的目标是尽可能快地多次执行一些代码并将结果存储在某处（总执行量将至少为 100M 次）。

逻辑看起来像这样（它是一个简化的例子）：

dbconn = new dbconn() //This is reused by all threads
for a in listOfSize1000:
   for b in listofSize10:
      for c in listOfSize2:
         taskcompletionexecutorservice.submit(new runner(a, b, c, dbconn))

最后，taskcompletionexecutorservice.take() 被调用，我将“未来”的结果存储在数据库中。 但这种方法并没有真正在一个点之后进行缩放。

这就是我现在在 spark 中所做的事情（这是一个残酷的 hack，但我正在寻找有关如何最好地构建它的建议）：

sparkContext.parallelize(listOfSize1000).filter(a -> {
   dbconn = new dbconn() //Cannot init it outsize parallelize since its not serializable
   for b in listofSize10:
      for c in listOfSize2:
         Result r = new runner(a, b, c. dbconn))
         dbconn.store(r)

    return true //It serves no purpose.
}).count();

这种方法在我看来效率低下，因为它并没有真正在最小的工作单元上进行并行化，尽管这项工作工作正常。另外 count 并没有真正为我做任何事情，我添加它来触发执行。它的灵感来自计算此处的 pi 示例：http://spark.apache.org/examples.html

那么对于如何更好地构建我的 spark runner 以便我可以有效地使用 spark executor 有什么建议吗？

Answer 1

所以我们可以做一些事情来让这段代码更像 Spark。首先是您使用了filter 和count，但实际上使用了两者的结果。函数foreach 可能更接近你想要的。

话虽如此，您正在创建一个数据库连接来存储结果，我们可以通过几种方式来实现这一点。一个是：如果数据库真的是您想要用于存储的东西，您可以使用foreachPartition 或mapPartitionsWithIndex 为每个分区只创建一个连接，然后执行count()（我知道这有点难看，但@ 987654328@ 自 1.0.0 起已弃用）。您也可以只做一个简单的map，然后将您的结果保存为许多支持的输出格式之一（例如 saveAsSequenceFile）。

Answer 2

您可以尝试另一种方法来更好地并行化它，但要付出代价。代码在 scala 中，但 python 有 cartesian 方法。为简单起见，我的列表包含整数。

val rdd1000 = sc.parallelize(list1000)
val rdd10 = sc.parallelize(list10)
val rdd2 = sc.parallelize(list2)

rdd1000.cartesian(rdd10).cartesian(rdd2)
    .foreachPartition((tuples: Iterator[Tuple2[Tuple2[Int, Int], Int]]) => {
        dbconn =...
        for (tuple <- tuples) {
            val a = tuple._1._1
            val b = tuple._1._2
            val c = tuple._2

            val r = new Result(a, b, c, dbconn)
            dbconn.store(r)
        }
    })

Filter 在您的情况下是一个 transformation，它是惰性的 - spark 不会在调用时对其进行评估。该过程仅在调用操作时开始。 Collect 是一个 action，它在您的示例中开始实际处理。 ForeachPartition 也是一个动作，spark 会立即启动它。 ForeachPartition 在这里是必需的，因为它允许为数据的整个分区打开一次连接。

笛卡尔的坏处可能是它可能意味着对集群进行洗牌，所以如果你有复杂的对象，那可能会影响性能。如果您要从外部源读取数据，则可能会发生这种情况。如果您要使用并行化，那很好。

还有一点需要注意的是，根据集群 Spark 的大小，可能会对您使用的数据库造成相当大的压力。