"Conteneur tué par YARN pour avoir dépassé les limites de mémoire. 10,4 Go de mémoire physique 10,4 Go utilisés "sur un cluster EMR avec 75 Go de mémoire

Je suis en cours d'exécution d'un nœud 5 Étincelle de cluster sur AWS DME chacun de taille m3.xlarge (1 master 4 esclaves). J'ai réussi a couru à travers une 146Mb bzip2 compressé fichier CSV, et qui finit avec une parfaite agrégées résultat.

Maintenant je suis en train de traiter environ 5 go de bzip2 fichier CSV sur ce groupe mais je reçois cette erreur:

16/11/23 17:29:53 AVERTIR TaskSetManager: Perdu tâche à 49,2 en scène 6.0 (TID xxx, xxx.xxx.xxx.de calcul.interne): ExecutorLostFailure (exécuteur 16 quitté provoquée par l'une des tâches en cours d'exécution) Raison: Conteneur tué par le FILS pour le dépassement des limites de la mémoire. 10.4 GO de 10,4 GO de mémoire physique utilisée. Envisager de multiplier étincelle.de fil.exécuteur testamentaire.memoryOverhead.

Je suis confus quant à pourquoi je suis un ~10.5 GB limite de mémoire sur un ~75 GO de cluster (15GO par 3m.xlarge exemple)...

Voici mon DME config:

[
 {
  "classification":"spark-env",
  "properties":{

  },
  "configurations":[
     {
        "classification":"export",
        "properties":{
           "PYSPARK_PYTHON":"python34"
        },
        "configurations":[

        ]
     }
  ]
},
{
  "classification":"spark",
  "properties":{
     "maximizeResourceAllocation":"true"
  },
  "configurations":[

  ]
 }
]

De ce que j'ai lu, la définition de la maximizeResourceAllocation propriété doit dire EMR pour configurer Étincelle d'utiliser pleinement toutes les ressources disponibles sur le cluster. C'est à dire, je devrais avoir ~75 GO de mémoire disponible... Alors pourquoi je reçois un ~10.5 GB de mémoire erreur de limite?
Voici le code, je suis en cours d'exécution:

def sessionize(raw_data, timeout):
# https://www.dataiku.com/learn/guide/code/reshaping_data/sessionization.html
    window = (pyspark.sql.Window.partitionBy("user_id", "site_id")
              .orderBy("timestamp"))
    diff = (pyspark.sql.functions.lag(raw_data.timestamp, 1)
            .over(window))
    time_diff = (raw_data.withColumn("time_diff", raw_data.timestamp - diff)
                 .withColumn("new_session", pyspark.sql.functions.when(pyspark.sql.functions.col("time_diff") >= timeout.seconds, 1).otherwise(0)))
    window = (pyspark.sql.Window.partitionBy("user_id", "site_id")
              .orderBy("timestamp")
              .rowsBetween(-1, 0))
    sessions = (time_diff.withColumn("session_id", pyspark.sql.functions.concat_ws("_", "user_id", "site_id", pyspark.sql.functions.sum("new_session").over(window))))
    return sessions
def aggregate_sessions(sessions):
    median = pyspark.sql.functions.udf(lambda x: statistics.median(x))
    aggregated = sessions.groupBy(pyspark.sql.functions.col("session_id")).agg(
        pyspark.sql.functions.first("site_id").alias("site_id"),
        pyspark.sql.functions.first("user_id").alias("user_id"),
        pyspark.sql.functions.count("id").alias("hits"),
        pyspark.sql.functions.min("timestamp").alias("start"),
        pyspark.sql.functions.max("timestamp").alias("finish"),
        median(pyspark.sql.functions.collect_list("foo")).alias("foo"),
    )
    return aggregated
 spark_context = pyspark.SparkContext(appName="process-raw-data")
spark_session = pyspark.sql.SparkSession(spark_context)
raw_data = spark_session.read.csv(sys.argv[1],
                                  header=True,
                                  inferSchema=True)
# Windowing doesn't seem to play nicely with TimestampTypes.
#
# Should be able to do this within the ``spark.read.csv`` call, I'd
# think. Need to look into it.
convert_to_unix = pyspark.sql.functions.udf(lambda s: arrow.get(s).timestamp)
raw_data = raw_data.withColumn("timestamp",
                               convert_to_unix(pyspark.sql.functions.col("timestamp")))
sessions = sessionize(raw_data, SESSION_TIMEOUT)
aggregated = aggregate_sessions(sessions)
aggregated.foreach(save_session)

Fondamentalement, rien de plus que de fenêtrage et un groupBy pour agréger les données.

Il commence avec quelques-unes de ces erreurs, et en direction de l'arrêt de la hausse du montant de la même erreur.

J'ai essayé de courir étincelle soumettre avec --conf étincelle.de fil.exécuteur testamentaire.memoryOverhead mais qui ne semble pas résoudre le problème.

source d'informationauteur lauri108

  1. 21

    Je ressens votre douleur..

    Nous avons eu des problèmes similaires de manquer de mémoire avec une Étincelle sur le FILS. Nous avons cinq de 64 go, 16 core VMs et indépendamment de ce que nous avons mis spark.yarn.executor.memoryOverheadnous ne pouvions tout simplement pas obtenir assez de mémoire pour ces tâches -- ils finiraient par mourir n'importe comment beaucoup de mémoire, nous leur donnerait. Et ce relativement simple Étincelle application qui a été la cause de cette fermeture.

    Nous en avons déduit que l'utilisation de la mémoire physique était assez faible sur la Vm, mais l'utilisation de la mémoire virtuelle a été extrêmement élevé. Nous avons mis yarn.nodemanager.vmem-check-enabled dans yarn-site.xml à false et nos conteneurs n'ont plus tué, et l'application semble fonctionner comme prévu.

    Faire plus de recherches, j'ai trouvé la réponse à pourquoi ça se passe ici: https://www.mapr.com/blog/best-practices-yarn-resource-management

    Depuis sur Centos/RHEL 6 est agressif allocation de mémoire virtuelle en raison d'OS comportement, vous devez désactiver la mémoire virtuelle vérificateur ou de l'augmentation de la laine.nodemanager.vmem-pmem-ratio à une relativement plus grande valeur.

    Que la page a un lien vers un très utile de la page d'IBM: https://www.ibm.com/developerworks/community/blogs/kevgrig/entry/linux_glibc_2_10_rhel_6_malloc_may_show_excessive_virtual_memory_usage?lang=en

    En résumé, la glibc > 2.10 a changé son allocation de la mémoire. Et bien que d'énormes quantités de mémoire virtuelle allouée n'est pas la fin du monde, il ne fonctionne pas avec les paramètres par défaut de FIL.

    Au lieu de yarn.nodemanager.vmem-check-enabled à false, vous pouvez également jouer avec le réglage de la MALLOC_ARENA_MAX variable d'environnement à un faible nombre de hadoop-env.sh.

    Je vous recommande la lecture par le biais de deux pages, l'information est très pratique.

  2. 4

    Si vous ne l'utilisez pas spark-submitet vous êtes à la recherche d'une autre façon de spécifier la yarn.nodemanager.vmem-check-enabled paramètre mentionné par Duffvoici 2 autres façons:

    Méthode 2

    Si vous utilisez un fichier de Configuration JSON (que vous passez à la AWS de la CLI ou de votre boto3 script), vous devrez ajouter la configuration suivante:

    [{
"Classification": "yarn-site", 
  "Properties": {
    "yarn.nodemanager.vmem-check-enabled": "false"
   }
}]

    Méthode 3

    Si vous utilisez le DME de la console, ajoutez la configuration suivante:

    classification=yarn-site,properties=[yarn.nodemanager.vmem-check-enabled=false]
  3. 3

    Voir,

    J'ai eu le même problème en un gigantesque amas que je suis aujourd'hui. Le problème ne sera pas résolu pour ajouter de la mémoire à l'ouvrier. Parfois, dans le processus d'agrégation étincelle va utiliser plus de mémoire qu'il a et l'étincelle emplois d'utiliser de la mémoire du tas.

    Un exemple simple est:

    Si vous avez un ensemble de données dont vous avez besoin pour reduceByKey il sera, parfois, agréger plus de données dans un seul travailleur que les autres, et si ces données exeeds la mémoire de l'un travailleur, vous obtenez ce message d'erreur.

    L'ajout de l'option spark.yarn.executor.memoryOverhead vous aidera si vous avez mis à 50% de la mémoire utilisée pour le travailleur (juste pour tester et voir si cela fonctionne, vous pouvez ajouter du moins avec plus de tests).

    Mais vous avez besoin de comprendre comment Spark travaille avec l'Allocation de Mémoire dans le cluster:

    1. La façon la plus commune de l'Étincelle utilise 75% de la mémoire de la machine. Le reste va à l'.
    2. Étincelle a deux types de la mémoire lors de l'exécution. Une partie de l'exécution et de l'autre est le stockage. L'exécution est utilisé pour le Mélange, les Jointures, les Agrégations et les Etc. Le stockage est utilisé pour la mise en cache et de la propagation de données à travers le cluster.

    Une bonne chose à propos de l'allocation de mémoire, si vous n'êtes pas à l'aide de cache dans votre exécution, vous pouvez définir l'étincelle pour l'utilisation que sotorage espace de travail avec exécution de façon à éviter en partie le OOM erreur. Comme vous pouvez le voir dans la documentation de l'étincelle:

    Cette conception assure plusieurs propriétés désirables. Tout d'abord, les applications qui n'utilisent pas de mise en cache permet d'utiliser l'ensemble de l'espace pour l'exécution, ce qui évite disque inutile de déversements. Deuxièmement, les applications qui utilisent le système de cache peut réserver un minimum d'espace de stockage (R) où leurs blocs de données sont à l'abri d'être expulsés. Enfin, cette approche offre raisonnable dehors-de-le-boîte de performances pour une variété de charges de travail sans exiger de l'utilisateur de l'expertise de la façon dont la mémoire est divisée en interne.

    Mais comment pouvons-nous l'utiliser?

    Vous pouvez modifier certaines configurations, Ajouter le MemoryOverhead configuration de votre emploi appel, mais, envisager d'ajouter: spark.memory.fraction changement de 0,8 ou de 0,85 et de réduire le spark.memory.storageFraction ou de 0,2 à 0,35.

    D'autres configurations peuvent aider, mais il faut vérifier dans votre cas. Se tous ces configuration ici.

    Maintenant, ce qui m'aide dans Mon cas.

    J'ai un cluster avec 2,5 K travailleurs et de 2,5 to de RAM. Et nous étions face à OOM erreur comme la vôtre. Nous venons d'augmenter la spark.yarn.executor.memoryOverhead à 2048. Et nous permettre la l'allocation dynamique. Et lorsque nous appelons le travail, nous n'avons pas l'ensemble de la mémoire pour les travailleurs, nous laissons cela pour l'Étincelle de décider. Nous venons de régler les frais Généraux.

    Mais pour certains tests pour mon petit cluster, la modification de la taille de l'exécution et de la mémoire de stockage. Qui a résolu le problème.