Spark formulaire de données formulaire

Question

J'ai fait un cluster apache autonome avec 7 pcs. Pour exécuter le code scala, le code est

---

/** Our main function where the action happens */ 

def main(args: Array[String]) { 

    // Set the log level to only print errors 

    Logger.getLogger("org").setLevel(Level.ERROR) 

    // Create a SparkContext without much actual configuration 

    // We want EMR's config defaults to be used. 

    val conf = new SparkConf() 

    conf.setAppName("MovieSimilarities1M") 

    val sc = new SparkContext(conf) 

    val input = sc.textFile("file:///home/ralfahad/LearnSpark/SBTCreate/customer-orders.csv") 

    val mappedInput = input.map(extractCustomerPricePairs) 

    val totalByCustomer = mappedInput.reduceByKey((x,y) => x + y) 

    val flipped = totalByCustomer.map(x => (x._2, x._1)) 

    val totalByCustomerSorted = flipped.sortByKey() 

    val results = totalByCustomerSorted.collect() 

    // Print the results. 

    results.foreach(println) 

    } 

} 

---

étapes sont:

1. Je créer un fichier .jar à l'aide de SBT

2. soumettre le travail à l'aide étincelles soumettre * .jar

Mais mon exécuteur testamentaire ne peut pas localiser sc.textFile("file:///home/ralfahad/LearnSpark/SBTCreate/customer-orders.csv")

Ce fichier client orders.csv est stocké dans mon PC maître.

stacktrace complète:

error: [Stage 0:> (0 + 2)/2]17/09/25 17:32:35 ERROR TaskSetManager: Task 0 in stage 0.0 failed 4 times; aborting job Exception in thread "main" org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 0.0 failed 4 times, most recent failure: Lost task 0.3 in stage 0.0 (TID 5, 141.225.166.191, executor 2): java.io.FileNotFoundException: File file:/home/ralfahad/LearnSpark/SBTCreate/customer-orders.csv does not exist

Comment puis-je résoudre ce problème?

Veuillez modifier le code à exécuter dans mon cluster.

Answer 1

Pour que vos nœuds de travail puissent accéder au fichier, vous avez plusieurs options.

1. Copiez manuellement le fichier sur tous les noeuds.

Chaque nœud doit avoir ce fichier exactement ce chemin: /home/ralfahad/LearnSpark/SBTCreate/customer-orders.csv

2. Envoyez votre travail avec le fichier joint.

spark-submit a un appel d'option --files qui vous permet de copier un certain nombre de fichiers tout en soumettant le travail comme celui-ci:

spark-submit --master ... -jars ... --files /home/ralfahad/LearnSpark/SBTCreate/customer-orders.csv 

Ne pas abuser. Cette option est plus destinée aux tests et aux petits fichiers.

3. Utilisez un espace de stockage externe disponible universellement auquel tous les noeuds peuvent accéder.

Les partages S3 et NFS sont des choix populaires.

sc.textFile("s3n://bucketname/customer-orders.csv") 

4. Vous pouvez lire les données en vous programme pilote, puis le convertir en faire RDD pour le traitement.

val bufferedSource = io.Source.fromFile("/home/ralfahad/LearnSpark/SBTCreate/customer-orders.csv") 
val lines = (for (line <- bufferedSource.getLines()) yield line).toList 
val rdd = sc.makeRdd(lines) 

généralement pas recommandé, mais peut être utilisé pour des tests rapides.