Spark中自定义对象的RDD问题

Question

我像这样将一个类型传递给平面地图；

 val customData: RDD[(Text/String, Custom)] = origRDD.flatMap { case(x) => parse(x)}

这会返回一个String和Custom的键值对（我在String上也使用了Text，但无济于事，这就是'Text/String'的解释）。这个类扩展了 Serializable 并在 Kryo 中注册。

当我尝试运行该程序时，它只是运行并且永不结束。永不结束是指我已经让它运行了 18 小时，但它没有完成。如果我将其更改为带有 Int（计数器）而不是自定义对象的 Text(hadoop io)，它会很快完成。当我说快速时，我的意思是 30 分钟。它运行的数据是相同的数据，并且两者都使用了 parse（以及 flatmap）方法，因此它运行的是完全相同的逻辑。它在 flatmap 中使用的方法是相同的 parse 方法。当我将其从 (Text, Int) 更改为 (Text/String, Custom) 时，行为会下降。

我想知道我需要添加什么才能完成这项工作。它需要是可写的吗？

自定义对象类实现示例（显然不准确但很好模仿）；

class Custom(dateAsLong: java.lang.Long, typeOfTransaction: util.HashSet[java.lang.Long], isCustomer: Boolean, amount: String, customerMatch: ObjectMatch) extends Serializable {
//has getters and setters here 

 val startDate = dateAsLong
 val transType = typeOfTransaction
 val customer = isCustomer
 val cost = amount
 val matchedData = customerMatch

 def getStartDate(): java.lang.Long = startDate
 def getTransType(): util.HashSet[java.lang.Long] = transType
 def getCustomer(): Boolean = customer
 def getCost(): String = amount
 def getMatchedData(): ObjectMatch = matchedData
}

扩展 java Serializable 的对象内部的 parse 方法示例；

object Paser extends Serializable { 
    def parse(transaction: Transaction, customerList: util.HashMap[String, String], storeList: util.HashMap[String, String]): List[(Text, Custom)] ={ //list because flatmap emits 0, 1 or 2 
//adds items to the list depending on conditions
    var list:List[(Text, Custom)] = List()
    val typeOfTransaction = getType(transaction)
    val dateAsLong = getDate(transaction)
    val amount = getAmount(transaction)
    val customerMatch = getCustomerMatch(transaction, customerList)
    val storeMatch = getStoreMatch(transaction, storeList)
     //more fields parsed

    if (customerMatch != Some(null)){
       isCustomer = true
       val transaction: Custom = getTransaction(dateAsLong, typeOfTransaction,      isCustomer, amount, customerMatch)
       val transactionId = hash(transaction.getCustomer(), transaction.getTransType(), transaction.getMatchedData().getItem())
       list = list :+ (new Text(transactionId), transaction)

    }  
    if (storeMatch != Some(null)){
       isCustomer = false
       val transaction: Custom = getTransaction(dateAsLong, typeOfTransaction,      isCustomer, typeOfTransaction, storeMatch)
       val transactionId = hash(transaction.getCustomer(), transaction.getTransType(), transaction.getMatchedData().getItem())
       list = list :+ (new Text(transactionId), transaction)
    }
  }
 list
}

Kryo 序列化是这样的；

 conf.registerKryoClasses(Array(classOf[Custom]))

感谢您提供代码示例或示例链接的任何帮助。

用于（文本/字符串、自定义）运行的 Spark UI

1/11任务的底部进度是平面图，顶部是saveAsNewHadoopAPIFile

平面图阶段 0 是 saveAsNewHadoopAPIFile -> filter x7 -> flatmap

使用 (Text, Int) 运行

慢速运行（文本/字符串，自定义）运行时间为 1.1 小时，但我让它运行了 18 小时。当它运行 18 小时时，它进展缓慢，但是让它运行一天并不理想。有什么不对劲，非常不对劲。再次在两者中使用 parse 方法，因此它运行完全相同的逻辑，即使更快的运行不输出自定义值，而是输出键 Text 和 int。

不确定这是否有帮助，但无论是什么错误都会导致 Accumulo 中的扫描看起来也有所不同。 Text, Int 运行的扫描数正常增加，在 30 分钟内保持相对相同的扫描数/秒，然后下降。当我使用自定义运行时，它会增加然后立即下降。以类似 ^ 的方式，然后以较低的扫描速率持续数小时。

Spark 版本：1.6.2，Scala 2.11

Answer 1

您不应使用var list:List[(Text, Custom)] = List()。每次执行代码list = list :+ (new Text(transactionId), transaction) 都会创建一个新列表（它不仅仅是添加到现有列表中）。 Scala 中的List 是不可变的。您应该将其替换为val myListBuffer = ListBuffer[(Text, Custom)]()。我不确定这是否是唯一的问题 - 但如果您的列表很大，此更改应该会有所帮助。

此外，还有一些关于在 Scala 中编写代码的 cmets - Scala 类中不需要有 getter 和 setter。无论如何，所有成员都是不可变的。在 Scala 中使用 var 之前，您需要认真思考。不变性将使您的代码具有弹性，提高可读性并且更容易更改。