Aléatoire désagrégé

Le service de mélange désagrégé permet à la majeure partie des données de mélange de résider dans un emplacement de stockage distinct (montage de volume partagé distinct ou librairie) à partir des noeuds de traitement, ce qui permet une utilisation efficace des ressources. Le plug-in Disaggregated shuffle utilise l'API Spark shuffle manager qui remplace le gestionnaire Shuffle existant. Il enregistre un nouveau composant ShuffleManager et ShuffleDataIO qui permet au plugin de surcharger les méthodes d'E/S Spark shuffle.

Le plug-in Disaggregated shuffle est utilisé par les travaux Spark qui sont à court de stockage temporaire local ou qui connaissent des échecs de pod. Il est également utilisé lorsque les ressources de stockage locales sont insuffisantes ou comme alternative à emptyDir pour améliorer les performances. Le plug-in Disaggregated shuffle prend en charge les stockages suivants:

• Volumes partagés Cloud Pak for Data
• Tout stockage compatible Spark tel que cos://bucket, s3a://bucket, hdfs: // ou file:///path

1. Sauvegardez l'exemple de fichier Python suivant.

   from pyspark.sql import SparkSession
   import time
   
   def init_spark():
   
       spark = SparkSession.builder.appName("auto-scale-test").getOrCreate()
   
       sc = spark.sparkContext
   
       return spark, sc
   
   def main():
   
       spark, sc = init_spark()
   
       partitions = [10, 5, 8, 4, 9, 4, 7]
   
       parallelism = sc.defaultParallelism
   
       for num_partitions in partitions:
   
           print(
   
               f"Running experiment with {num_partitions} partitions leveraging {parallelism} cores."
   
           )
   
           data = range(1, 20000000)
   
           v0 = sc.parallelize(data, parallelism)
   
           gr = v0.groupBy(lambda x: x % num_partitions, numPartitions=num_partitions).map(
   
               lambda x: (x[0], len(x[1]))
   
           )
   
           dd = gr.collect()
   
           print(f"Buckets {dd}")
   
           time.sleep(5)
   
   if __name__ == "__main__":
   
       main()
   

2. Téléchargez le fichier Python dans le compartiment Cloud Object Storage .

3. Pour activer la fonction de mélange sur un volume partagé dans Analytics Engine, vous devez monter le volume partagé et spécifier le chemin dans la fonction de mélange désagrégé. Exécutez la commande curl suivante pour soumettre l'application Spark.

curl -k -X POST https://<cluster-url>/v4/analytics_engines/<app-id>/spark_applications \
    -H "Authorization: ZenApiKey <api-key>" \
    -d '{  "application_details": {\
    "conf": {\
           "spark.hadoop.fs.s3a.endpoint": "s3.us-south.cloud-object-storage.appdomain.cloud",\
           "spark.hadoop.fs.s3a.access.key": "<access-key>",\
           "spark.hadoop.fs.s3a.secret.key": "<secret-key>",\
           "spark.hadoop.fs.s3a.path.style.access": "true",\
           "spark.hadoop.fs.s3a.fast.upload": "true","spark.hadoop.fs.s3a.impl": "org.apache.hadoop.fs.s3a.S3AFileSystem",\
           "spark.shuffle.manager": "org.apache.spark.shuffle.sort.DisaggregatedShuffleManager",\
           "spark.shuffle.sort.io.plugin.class": "org.apache.spark.shuffle.DisaggregatedShuffleDataIO",\
           "spark.shuffle.disaggregated.rootDir": "file:///<path-name>/",\
           "spark.shuffle.disaggregated.folderPrefixes": "1"    },\
           "application": "s3a://<cos-bucket-name>/<app-name>.py"  },
            "volumes": [{    "name": "cpd-instance::<volume-name>",
            "mount_path": "file:///<path-name>"  }]}'

Le plug-in nettoie automatiquement les fichiers shuffle créés. Cependant, les fichiers de remaniement créés lorsqu'un pilote Spark est arrêté ou que des pannes ne sont pas removed.It de nettoyer régulièrement le répertoire de remaniement. Si les fichiers shuffle ont une sauvegarde de librairie, la règle de cycle de vie peut être implémentée sur le compartiment.