Ce qui est une tâche Spark? Comment l'Étincelle travailleur d'exécuter le fichier jar?

Après la lecture d'un document sur http://spark.apache.org/docs/0.8.0/cluster-overview.html, j'ai quelques question que je tiens à préciser.

Prendre cet exemple de l'Étincelle:

JavaSparkContext spark = new JavaSparkContext(
  new SparkConf().setJars("...").setSparkHome....);
JavaRDD<String> file = spark.textFile("hdfs://...");

//step1
JavaRDD<String> words =
  file.flatMap(new FlatMapFunction<String, String>() {
    public Iterable<String> call(String s) {
      return Arrays.asList(s.split(" "));
    }
  });

//step2
JavaPairRDD<String, Integer> pairs =
  words.map(new PairFunction<String, String, Integer>() {
    public Tuple2<String, Integer> call(String s) {
      return new Tuple2<String, Integer>(s, 1);
    }
  });

//step3
JavaPairRDD<String, Integer> counts =
  pairs.reduceByKey(new Function2<Integer, Integer>() {
    public Integer call(Integer a, Integer b) {
      return a + b;
    }
  });

counts.saveAsTextFile("hdfs://...");

Donc, disons que j'ai 3 nœuds de cluster, et le nœud 1 exécute en tant que maître, et au-dessus de programme pilote a été correctement jared (dire application-test.jar). Alors maintenant, je suis en cours d'exécution de ce code sur le nœud maître et je crois que juste après la SparkContext en cours de création, l'application-test.jar le fichier sera copié dans les nœuds de travail (et de chaque travailleur va créer un dir pour cette application).

Alors maintenant ma question:
Sont etape 1, etape 2 et etape 3 dans l'exemple des tâches qui sont envoyés sur les travailleurs? Si oui, alors comment le travailleur doit-il exécuter? Comme java -cp "application-test.jar" step1 et ainsi de suite?

InformationsquelleAutor EdwinGuo | 2014-08-13

apache-spark distributed-computing

96

Lorsque vous créez le SparkContext, chaque travailleur commence un exécuteur. C'est un processus distinct (JVM), et elle charge votre bocal de trop. Les exécuteurs testamentaires se connecter de nouveau à votre programme de pilote. Maintenant, le conducteur peut envoyer les commandes, comme flatMap, map et reduceByKey dans votre exemple. Lorsque le conducteur s'arrête, les exécuteurs testamentaires de l'arrêter.

Rdd sont un peu comme les grands tableaux qui sont divisés en partitions, et chaque interprète peut tenir certaines de ces partitions.

Un tâche est une commande envoyée par le conducteur d'un exécuteur testamentaire par la sérialisation de votre Function objet. L'exécuteur désérialise la commande (ce qui est possible parce qu'il a chargé votre jar), et l'exécute sur une partition.

_{(C'est une vue d'ensemble conceptuelle. Je suis abstraction de certains détails, mais j'espère que c'est utile.)}

Pour répondre à votre question: Non, un nouveau processus n'est pas lancé pour chaque étape. Un nouveau processus est démarré sur chaque travailleur lors de la SparkContext est construit.
- Merci Daniel! vraiment vous remercie de votre réponse, très utile!! Couple de choses que je tiens à préciser: 1) de sorte que lorsque la tâche est exécutée sur le travailleur, d'exécuteur testamentaire, d'utiliser le pot comme une dépendance à exécuter? 2) quand vous dites stade, je peux comprendre etape 1 etape 2.. dans mon exemple? 3) est une tâche égale à une fonction qui fait appel à la rdd les fonctions de transformation, à l'instar de la carte et flatMap?
- 1) Oui. 2) Oui. J'ai fixé ma réponse à-dire pas au lieu de stade. 3) Oui, assez bien. Si vous exécutez rdd.map(myFunc), une tâche est créée pour chaque partition. Les exécuteurs testamentaires de ramassage et d'exécuter les tâches. Chaque tâche dans ce cas va myFunc sur une partition. Certaines opérations, comme reduceByKey sont plus complexes, mais c'est l'idée de base.
- Il y a une nuance importante. Tout se passe dans un lazy moyen. Donc rdd.map ne fait rien jusqu'à ce qu'il en a besoin. Si vous ne rdd.filter(...).map(...).collect(), le filter et map fonctions de s'exécuter uniquement sur les travailleurs lorsque vous appelez collect. Mais la plupart du temps vous n'avez pas besoin de penser à ce sujet.
- donc, le exécuteurs fait pile la rdd tâche de transformation et de ne pas les exécuter, à moins qu'une action appelée sur le ca. c'est pourquoi il appelle la résilience? il enregistre la tâche de transformation et quand certains échoue, exécuteur il suffit de sélectionner la tâche et de courir à nouveau? Merci
- Exactement comme vous le dites.
- J'ai récemment découvert que je devais définir anounymous fonction au sein de la rdd les fonctions de transformation, ou j'ai besoin de définir une méthode statique dans une singlton objet. Je reviens donc ici et de prendre un second, bien que pour ce que vous m'avez dit, lorsque vous avez dit " C'est un processus distinct (JVM), et elle charge votre bocal de trop. Les exécuteurs testamentaires se connecter de nouveau à votre programme de pilote",
- si le fichier jar réellement battu pour le travailleur nœud? ou rester au niveau du pilote de l'application du nœud, et lorsque le pilote d'envoyer la tâche de l'ouvrier, et le travailleur de revenir à la pilote de référence et le pot comme la dépendance à l'exécution de la commande? donc, pendant toute la durée du calcul, il n'y a qu'une seule copie du fichier jar? et il ne reste au pilote de nœud? Merci
- Le pot que vous spécifiez avec SparkContext.addJar sera copié sur tous les nœuds du travailleur.
- C'est comme une ... belle ... la narration des concepts de base de la bougie d'allumage qui il m'a conduit presque les larmes aux yeux. Peut-être ce qui est trompeur, c'est le concept que les RDD sont comme de grands tableaux. Ils sont un récipient contenant des instructions sur la façon de matérialiser ces tableaux, et la façon de partition, pas de ceux des tableaux eux-mêmes.
- Hey Daniel, le code qui lit depuis un fichier texte (ou S3), une prise réseau) également exécuter à l'intérieur de l'Exécuteur testamentaire?
- Oui. Si vous utilisez sc.textFile, le pilote de l'application sera d'abord obtenir la liste de "divisions" (des morceaux du fichier) à partir du système de fichiers. Puis l'exécuteur fils de chaque charge d'un split à un moment en parallèle.
- Lorsque DataFrameWriter.enregistrer(une Chaîne-s3-chemin) est invoquée, c'est que s'exécuter à l'intérieur d'exécuteurs? Si non, alors le maître sera probablement passer beaucoup de temps à écrire le fichier en lui-même si le chemin est un chemin d'accès réseau...avant de continuer. Est ce que le droit? Si oui, exécuteurs pas tenter de neutraliser les uns des autres données?
- sera la cause de la exécuteurs pour écrire les données dans un répertoire donné. Chaque partition est écrite dans un fichier séparé, donc il n'y a pas de problème de collision. Le correspondant load méthode peut charger à partir d'un répertoire, donc le fait que chaque partition est un fichier distinct est la plupart du temps cachés.
InformationsquelleAutor Daniel Darabos
32

Pour obtenir un clair aperçu de la façon dont les tâches sont créés et programmés, nous devons comprendre comment le modèle d'exécution de travaux dans Spark. Peu de temps de parler, une application spark est réalisée en trois étapes :
1. Créer des RDD graphique
2. Créer plan d'exécution en fonction de la RDD graphique. Les étapes sont créés dans cette étape
3. Générer des tâches en fonction du plan et de les faire planifiée à travers travailleurs
De votre mot-comte exemple, le RDD graphique est plutôt simple, c'est quelque chose comme suit :

fichier -> lignes -> mots -> par le nombre de mots -> global nombre de mots -> sortie

Basé sur ce graphique, deux étapes sont créés. L'étape de création de la règle est fondée sur l'idée de pipeline nombreuses et étroites transformations que possible. Dans votre exemple, l'étroitesse de la transformation se termine à chaque mot compte. Par conséquent, vous obtenez deux étapes
1. fichier -> lignes -> mots -> par le nombre de mots
2. mondiale nombre de mots -> sortie
Une fois que les étapes sont compris, étincelle va générer des tâches à partir d'étapes. La première étape sera de créer ShuffleMapTasks et la dernière étape sera de créer ResultTasks parce que dans la dernière étape, une action de l'opération est inclus afin de produire des résultats.

Le nombre de tâches à être généré dépend de la façon dont vos fichiers sont distribués. Supposons que vous ayez 3 trois fichiers différents dans trois différents nœuds, la première étape sera de générer des 3 tâches : une tâche par partition.

Par conséquent, vous ne devriez pas la carte à vos mesures pour les tâches directement. Une tâche appartient à une étape, et est liée à une partition.

Généralement, le nombre de tâches a couru pour un stade est exactement le nombre de partitions de la finale de la RDD, mais depuis Rdd peuvent être partagées (et donc ShuffleMapStages) leur nombre varie en fonction de la RDD/stade de partage. Veuillez vous référer à Comment DAG fonctionne sous les couvertures dans les RDD?
- Grâce Hui . Il un sens aujourd'hui pourquoi j'ai près de 20 tâches pour une seule scène. Mon CA est a 20 partitions sur cluster. Une question cependant, comment puis-je la force de chaque exécuteur de travailler seulement sur des données locales. Je vois que l'exécuteur des journaux en disant: "de stockage.La commande gestionnaire de blocs: Trouvé bloc rdd_2_2 à distance". La plupart du bloc semblait être trouvé local mais certains sont marqués comme à distance. Quand j'observe les étapes de chaque sur est soit NODE_LOCAL ou PROCESS_LOCAL
- je ne connais pas tous les détails sur la façon d'exécuteur des œuvres, mais à quoi vous faites allusion peut-être au cas où des données doivent être mélangées : déplacement d'un nœud à un autre nœud. Si oui, vous ne pouvez pas le forcer car il n'a besoin de ces données à distance.
- Je pense par RDD graphique vous dire le groupe graphique.
InformationsquelleAutor Hui Wang

Vous devez vous connecter pour publier un commentaire.