MaxCompute:Utiliser des volumes externes pour traiter des données non structurées

Utiliser Spark on MaxCompute avec des volumes externes

Spark on MaxCompute est un service de calcul fourni par MaxCompute, compatible avec Spark open source. Il offre un framework de calcul Spark basé sur un système unifié pour les ressources de calcul, les jeux de données et les autorisations. Cela vous permet de soumettre et d'exécuter des jobs Spark en utilisant des méthodes de développement familières pour répondre à une large gamme de besoins en matière de traitement et d'analyse des données. Si vous avez besoin d'un contrôle d'accès granulaire, utilisez un volume externe pour gérer l'accès via le système d'autorisations de l'entrepôt de données. Pour plus d'informations, consultez la rubrique Accéder à OSS depuis Spark on MaxCompute.

-- Configurations
spark.hadoop.odps.cupid.volume.paths=odps://ms_proj1_dev/volume_yyy1/
spark.hadoop.odps.volume.common.filesystem=true
spark.hadoop.fs.odps.impl=org.apache.hadoop.fs.aliyun.volume.OdpsVolumeFileSystem
spark.hadoop.fs.AbstractFileSystem.odps.impl=org.apache.hadoop.fs.aliyun.volume.abstractfsimpl.OdpsVolumeFs

spark.hadoop.odps.access.id=xxxxxxxxx
spark.hadoop.odps.access.key=xxxxxxxxx
spark.hadoop.fs.oss.endpoint=oss-cn-beijing-internal.aliyuncs.com
spark.hadoop.odps.cupid.resources=ms_proj1_dev.jindofs-sdk-3.8.0.jar
spark.hadoop.fs.oss.impl=com.aliyun.emr.fs.oss.JindoOssFileSystem

spark.hadoop.odps.cupid.resources=public.python-2.7.13-ucs4.tar.gz
spark.pyspark.python=./public.python-2.7.13-ucs4.tar.gz/python-2.7.13-ucs4/bin/python
spark.hadoop.odps.spark.version=spark-2.4.5-odps0.34.0

-- Code
from numpy import array
from math import sqrt

from pyspark import SparkContext
from pyspark.mllib.clustering import KMeans, KMeansModel

if __name__ == "__main__":
    sc = SparkContext(appName="KMeansExample")  # SparkContext

    # Load and parse the data
    data = sc.textFile("odps://ms_proj1_dev/volume_yyy1/kmeans_data.txt")
    parsedData = data.map(lambda line: array([float(x) for x in line.split(' ')]))

    # Build the model (cluster the data)
    clusters = KMeans.train(parsedData, 2, maxIterations=10, initializationMode="random")

    # Evaluate clustering by computing Within Set Sum of Squared Errors
    def error(point):
        center = clusters.centers[clusters.predict(point)]
        return sqrt(sum([x**2 for x in (point - center)]))

    WSSSE = parsedData.map(lambda point: error(point)).reduce(lambda x, y: x + y)
    print("Within Set Sum of Squared Error = " + str(WSSSE))

    # Save and load model
    clusters.save(sc, "odps://ms_proj1_dev/volume_yyy1/target/PythonKMeansExample/KMeansModel")

    print(parsedData.map(lambda feature: clusters.predict(feature)).collect())

    sameModel = KMeansModel.load(sc, "odps://ms_proj1_dev/volume_yyy1/target/PythonKMeansExample/KMeansModel")
    
    print(parsedData.map(lambda feature: sameModel.predict(feature)).collect())
    sc.stop()

Utiliser Proxima CE pour le calcul vectoriel

Cette section fournit des instructions et un exemple sur l'utilisation de Proxima CE pour la vectorisation dans MaxCompute.

1. Installez le package de ressources Proxima CE.

2. Exécutez un job.

   • Limitations :

     • Le SDK Proxima pour Java prend actuellement en charge l'exécution des commandes de tâches uniquement sur un client MaxCompute fonctionnant sous Linux ou macOS.

       Remarque

       Les jobs Proxima CE se composent de deux parties : les tâches locales et les tâches MaxCompute. Les tâches locales sont des modules fonctionnels qui n'impliquent pas de jobs SQL, MapReduce ou Graph MaxCompute. Les tâches MaxCompute sont des jobs exécutés par des moteurs MaxCompute tels que SQL, MapReduce et Graph. Ces deux types de tâches s'exécutent alternativement. Lorsqu'un job Proxima CE démarre, il tente d'abord de charger le noyau Proxima sur la machine locale où le job est exécuté, en utilisant le client MaxCompute. Si le chargement du noyau réussit, certains modules s'exécutent localement et appellent des fonctions basées sur le noyau Proxima. En cas d'échec du chargement du noyau, une erreur est signalée, mais les opérations suivantes ne sont pas affectées et les modules appellent des fonctions de secours à la place. Étant donné que le package JAR du job contient des dépendances liées à Linux, il ne peut pas être exécuté sur un client MaxCompute sous un système d'exploitation Windows.

     • L'exécution de jobs à partir de nœuds MapReduce DataWorks n'est pas prise en charge pour le moment. La version sous-jacente du client MaxCompute intégrée au nœud MapReduce est en cours de mise à niveau, ce qui entraîne l'échec des jobs. Pour l'instant, soumettez les jobs à l'aide du client MaxCompute.

   • Préparez les données :

     -- Create input tables.
     CREATE TABLE doc_table_float_smoke(pk STRING, vector STRING) PARTITIONED BY (pt STRING);
     CREATE TABLE query_table_float_smoke(pk STRING, vector STRING) PARTITIONED BY (pt STRING);
     
     -- Insert data into the doc table (base table).
     ALTER TABLE doc_table_float_smoke add PARTITION(pt='20230116');
     INSERT OVERWRITE TABLE doc_table_float_smoke PARTITION (pt='20230116') VALUES
     ('1.nid','1~1~1~1~1~1~1~1'),
     ('2.nid','2~2~2~2~2~2~2~2'),
     ('3.nid','3~3~3~3~3~3~3~3'),
     ('4.nid','4~4~4~4~4~4~4~4'),
     ('5.nid','5~5~5~5~5~5~5~5'),
     ('6.nid','6~6~6~6~6~6~6~6'),
     ('7.nid','7~7~7~7~7~7~7~7'),
     ('8.nid','8~8~8~8~8~8~8~8'),
     ('9.nid','9~9~9~9~9~9~9~9'),
     ('10.nid','10~10~10~10~10~10~10~10');
     
     -- Insert data into the query table.
     ALTER TABLE query_table_float_smoke add PARTITION(pt='20230116');
     INSERT OVERWRITE TABLE query_table_float_smoke PARTITION (pt='20230116') VALUES
     ('q1.nid','1~1~1~1~2~2~2~2'),
     ('q2.nid','4~4~4~4~3~3~3~3'),
     ('q3.nid','9~9~9~9~5~5~5~5');

   • Exemple de code de job :

     jar -libjars proxima-ce-aliyun-1.0.0.jar 
     -classpath proxima-ce-aliyun-1.0.0.jar com.alibaba.proxima2.ce.ProximaCERunner 
     -doc_table doc_table_float_smoke 
     -doc_table_partition 20230116 
     -query_table query_table_float_smoke 
     -query_table_partition 20230116 
     -output_table output_table_float_smoke 
     -output_table_partition 20230116 
     -data_type float 
     -dimension 8 
     -topk 1 
     -job_mode train:build:seek:recall 
     -external_volume shanghai_vol_ceshi
     -owner_id 1248953xxx
     ;

   • Exemple de résultat : exécutez la commande select * from output_table_float_smoke where pt='20230116'; pour interroger la table de résultats.

     +------------+------------+------------+------------+
     | pk         | knn_result | score      | pt         |
     +------------+------------+------------+------------+
     | q1.nid     | 2.nid      | 4.0        | 20230116   |
     | q1.nid     | 1.nid      | 4.0        | 20230116   |
     | q1.nid     | 3.nid      | 20.0       | 20230116   |
     | q2.nid     | 4.nid      | 4.0        | 20230116   |
     | q2.nid     | 3.nid      | 4.0        | 20230116   |
     | q2.nid     | 2.nid      | 20.0       | 20230116   |
     | q3.nid     | 7.nid      | 32.0       | 20230116   |
     | q3.nid     | 8.nid      | 40.0       | 20230116   |
     | q3.nid     | 6.nid      | 40.0       | 20230116   |
     +------------+------------+------------+------------+