Spark: не удается запустить терасор, когда количество данных становится больше.

Question

У меня есть искровая скамья, которая включает в себя терасор и работает нормально, когда данные составляют всего несколько сотен ГБ, но когда я генерирую больше данных, таких как 1 ТБ , пошло не так, в какой-то step.The следующий мой код:

import org.apache.spark.rdd._ 
import org.apache.spark._ 
import org.apache.spark.SparkContext._ 


object ScalaTeraSort{ 

    def main(args: Array[String]){ 
    if (args.length < 2){ 
     System.err.println(
     s"Usage: $ScalaTeraSort <INPUT_HDFS> <OUTPUT_HDFS>" 
    ) 
     System.exit(1) 
    } 
    val sparkConf = new SparkConf().setAppName("ScalaTeraSort") 
    val sc = new SparkContext(sparkConf) 

    val file = sc.textFile(args(0)) 
    val data = file.map(line => (line.substring(0, 10), line.substring(10))) 
        .sortByKey().map{case(k, v) => k + v} 
    data.saveAsTextFile(args(1)) 

    sc.stop() 
    } 

}

этот код в основном включает в себя 3 этапа: sortByKey, отображающие и saveAsTextFile. кажется, что в первых двух шагах нет ничего плохого, но когда дело доходит до третьего шага, он все время ошибался, а затем повторил второй шаг. третий шаг пошло не так из-за "FetchFailed (BlockManagerId (40, sr232, 44815, 0), shuffleId = 0, МАПИД = 11825, reduceId = 0)"

Answer 1

я узнал причину, существенное проблема заключается в: java.io.IOException: sendMessageReliably не удалось, потому что извед не был получен в течение 60 секунд

, который должен сказать, что вы должны установить свойство «spark.core.connection.ack.wait.timeout «к большему значению, по умолчанию это 60 секунд. Другие виды, этап не удастся из-за долгого времени, а не ответа.