如何使用Scala或Python在Apache Spark中运行多线程作业? [英] How to run Multi threaded jobs in apache spark using scala or python?

问题描述

我遇到了一个与火花并发有关的问题，这使我无法在生产中使用它，但我知道有一种解决方法.我正在尝试使用订单历史记录在700万用户上运行10亿种产品的Spark ALS.首先，我要列出不同的用户列表，然后在这些用户上运行循环以获取建议，这是一个非常缓慢的过程，并且需要几天的时间才能为所有用户获取建议.我尝试通过笛卡尔用户和产品来一次获得所有建议，但是再次将其提供给elasticsearch我不得不对每个用户的记录进行过滤和排序，然后才可以将其提供给Elasticsearch以供其他API使用.

I am facing a problem related to concurrency in spark which is stopping me from using it in production but I know there is a way out of it. I am trying to run Spark ALS on 7 million users for a billion products using order history. Firstly I am taking a list of distinct Users and then running a loop on these users to get recommendations, which is pretty slow process and will take days to get recommendations for all users. I tried doing cartesian users and products to get recommendations for all at once but again to feed this to elasticsearch I have to filter and sort records for each users and only then I can feed it to elasticsearch to be consumed by other APIs.

因此，请向我建议一种在这种用例中可扩展性很强的解决方案，并将其与实时建议一起用于生产中.

So please suggest me a solution which is pretty scalable in such use case and to be used in production with realtime recommendations.

这是我在scala中的代码段，它将使您了解我目前如何解决该问题:

Here is my code snippet in scala which will give you an idea how I am currently approaching to solve the problem:

  //    buy_values -> RDD with Rating(<int user_id>, <int product_id>, <double rating>)
  def recommend_for_user(user: Int): Unit = {
      println("Recommendations for User ID: " + user);
      // Product IDs which are not bought by user 
      val candidates = buys_values
        .filter(x => x("customer_id").toString.toInt != user)
        .map(x => x("product_id").toString.toInt)
        .distinct().map((user, _))
      // find 30 products with top rating
      val recommendations = bestModel.get
        .predict(candidates)
        .takeOrdered(30)(Ordering[Double].reverse.on(x => x.rating))

      var i = 1
      var ESMap = Map[String, String]()
      recommendations.foreach { r =>
        ESMap += r.product.toString -> bitem_ids.value(r.product)
      }
      //  push to elasticsearch with user as id
      client.execute {
        index into "recommendation" / "items" id user fields ESMap
      }.await
      // remove candidate RDD from memory
      candidates.unpersist()
  }
  // iterate on each user to get recommendations for the user [slow process]
  user_ids.foreach(recommend_for_user)

推荐答案

很明显，程序中的瓶颈是对candidates的搜索.有了Spark架构，它就严重限制了您的并行化能力，并且通过为每个用户启动Spark作业而增加了可观的开销.

It is pretty clear that bottleneck in your program is a search for candidates. Given the Spark architecture it severely limits your ability to parallelize and adds substantial overhead by starting Spark job for each user.

假设是典型情况，则在大多数情况下，拥有 700万用户和十亿种产品的用户可以预测整个产品范围，而少了用户已经购买的产品.至少在我看来，重要的问题是为什么还要打扰过滤.即使您推荐以前购买的产品，真的有害吗?

Assuming typical scenario, with 7 million users and a billion products most of time you'll predict over a whole range of products minus few already bought by the user. At least in my opinion important question is why even bother with filtering. Even if you recommend product which has been previously bought is it really harmful?

除非您有非常严格的要求，否则我将直接忽略该问题并使用

Unless you have very strict requirements I would simply ignore the problem and use MatrixFactorizationModel.recommendProductsForUsers which pretty much does all the job, excluding data export, for you. After that you can perform bulk export and you're good to go.

现在可以说您有一个明确的无重复政策.在假设典型用户仅购买相对较少数量的产品的前提下，您可以从为每个用户获取一组产品开始:

Now lets say you have a clear no-duplicates policy. Working under assumption that a typical user purchased only a relatively small number of products you can start with obtaining a set of products for each user:

val userProdSet = buy_values
    .map{case (user, product, _) => (user, product)} 
    .aggregateByKey(Set.empty[Int])((s, e) => s + e, (s1, s2) => s1 ++ s2)

接下来，您可以简单地映射userProdSet以获得预测:

Next you can simply map userProdSet to get predictions:

// Number of predictions for each user
val nPred = 30;

userProdSet.map{case (user, prodSet) => {
    val recommended = model
         // Find recommendations for user
        .recommendProducts(_, nPred + prodSet.size))
        // Filter to remove already purchased 
        .filter(rating => !prodSet.contains(rating.product))
        // Sort and limit
        .sortBy(_.rating)
        .reverse
        .take(nPred)
    (user, recommended)
}}

您可以通过使用可变集合进行聚合并广播模型来进一步改进，但这就是一个普遍的想法.

You can improve further by using mutable sets for aggregation and by broadcasting the model but thats a general idea.

如果user_ids中的用户数少于整个集合(buy_values)中的用户数，则只需过滤userProdSet以仅保留一部分用户.

If number of user in user_ids is lower than number of user in a whole set (buy_values) you can simply filter userProdSet to keep only a subset of users.

这篇关于如何使用Scala或Python在Apache Spark中运行多线程作业?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！