spark groupBy操作在199/200掛起

我有一個帶有master和兩個executors的spark單機羣。我有一個RDD[LevelOneOutput]及以下LevelOneOutput類spark groupBy操作在199/200掛起

class LevelOneOutput extends Serializable { 

    @BeanProperty 
    var userId: String = _ 

    @BeanProperty 
    var tenantId: String = _ 

    @BeanProperty 
    var rowCreatedMonth: Int = _ 

    @BeanProperty 
    var rowCreatedYear: Int = _ 

    @BeanProperty 
    var listType1: ArrayBuffer[TypeOne] = _ 

    @BeanProperty 
    var listType2: ArrayBuffer[TypeTwo] = _ 

    @BeanProperty 
    var listType3: ArrayBuffer[TypeThree] = _ 

    ... 
    ... 

    @BeanProperty 
    var listType18: ArrayBuffer[TypeEighteen] = _ 

    @BeanProperty 
    var groupbyKey: String = _ 
}

現在我想這組RDD基於用戶id，tenantId，rowCreatedMonth，rowCreatedYear。對於我這樣做

val levelOneRDD = inputRDD.map(row => { 
    row.setGroupbyKey(s"${row.getTenantId}_${row.getRowCreatedYear}_${row.getRowCreatedMonth}_${row.getUserId}") 
    row 
}) 

val groupedRDD = levelOneRDD.groupBy(row => row.getGroupbyKey)

這讓我在關鍵的數據作爲String和值Iterable[LevelOneOutput]

現在我想生成該組密鑰的LevelOneOutput一個單獨的對象。對於我在做類似如下：

val rdd = groupedRDD.map(row => { 
    val levelOneOutput = new LevelOneOutput 
    val groupKey = row._1.split("_") 

    levelOneOutput.setTenantId(groupKey(0)) 
    levelOneOutput.setRowCreatedYear(groupKey(1).toInt) 
    levelOneOutput.setRowCreatedMonth(groupKey(2).toInt) 
    levelOneOutput.setUserId(groupKey(3)) 

    var listType1 = new ArrayBuffer[TypeOne] 
    var listType2 = new ArrayBuffer[TypeTwo] 
    var listType3 = new ArrayBuffer[TypeThree] 
    ... 
    ... 
    var listType18 = new ArrayBuffer[TypeEighteen] 

    row._2.foreach(data => { 
    if (data.getListType1 != null) listType1 = listType1 ++ data.getListType1 
    if (data.getListType2 != null) listType2 = listType2 ++ data.getListType2 
    if (data.getListType3 != null) listType3 = listType3 ++ data.getListType3 
    ... 
    ... 
    if (data.getListType18 != null) listType18 = listType18 ++ data.getListType18 
    }) 

    if (listType1.isEmpty) levelOneOutput.setListType1(null) else levelOneOutput.setListType1(listType1) 
    if (listType2.isEmpty) levelOneOutput.setListType2(null) else levelOneOutput.setListType2(listType2) 
    if (listType3.isEmpty) levelOneOutput.setListType3(null) else levelOneOutput.setListType3(listType3) 
    ... 
    ... 
    if (listType18.isEmpty) levelOneOutput.setListType18(null) else levelOneOutput.setListType18(listType18) 

    levelOneOutput 
})

這爲預期輸入的小規模工作，但是當我嘗試在更大的一組輸入數據運行，由手術組是越來越掛在199/200，我沒有看到任何標準輸出特定錯誤或警告/標準錯誤

能有人指出我作業爲什麼沒有進一步繼續...

來源

2017-03-06 Prasad Khode

而不是使用groupBy操作，我創建配對RDD像低於

val levelOnePairedRDD = inputRDD.map(row => { 
    row.setGroupbyKey(s"${row.getTenantId}_${row.getRowCreatedYear}_${row.getRowCreatedMonth}_${row.getUserId}") 
    (row.getGroupByKey, row) 
})

並更新了處理邏輯，解決了我的問題。

來源

2017-08-05 07:39:36

spark groupBy操作在199/200掛起

回答

相關問題