MaxCompute:**Use volumes externos para processar dados não estruturados**

Use Spark on MaxCompute com volumes externos

O Spark on MaxCompute é um service de computação do MaxCompute compatível com o Spark open source. Ele oferece um framework de computação Spark sobre um sistema unificado de recursos de computação, conjuntos de dados e permissões. Isso permite enviar e executar jobs Spark usando métodos de desenvolvimento familiares para atender a diversas necessidades de processamento e análise de dados. Se for necessário controle de permissão granular, use um volume externo para gerenciar o acesso por meio do sistema de permissões do data warehouse. Para mais informações, consulte Access OSS from 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()

Use Proxima CE para computação vetorial

Esta seção fornece instruções e um exemplo de como usar o Proxima CE para vetorização no MaxCompute.

1. Instale o pacote de recursos do Proxima CE.

2. Execute um job.

   • Limitações:

     • Atualmente, o Proxima SDK for Java suporta apenas a execução de comandos de tarefa em um cliente MaxCompute rodando em Linux ou macOS.

       Nota

       Os jobs do Proxima CE consistem em duas partes: tarefas locais e tarefas do MaxCompute. Tarefas locais são módulos funcionais que não envolvem SQL do MaxCompute, MapReduce ou jobs Graph. Tarefas do MaxCompute são jobs executados por mecanismos do MaxCompute, como SQL, MapReduce e Graph. Esses dois tipos de tarefas são executados alternadamente. Quando um job do Proxima CE inicia, ele primeiro tenta carregar o kernel do Proxima na máquina local onde o job é executado, usando o cliente MaxCompute. Se o kernel carregar com sucesso, certos módulos rodam localmente e chamam funções baseadas no kernel do Proxima. Se o kernel falhar ao carregar, um erro é relatado, mas as operações subsequentes não são afetadas, e os módulos chamam funções de fallback. Como o pacote JAR do job contém dependências relacionadas ao Linux, ele não pode ser executado em um cliente MaxCompute em um sistema operacional Windows.

     • A execução de jobs a partir de nós MapReduce do DataWorks não é suportada atualmente. A versão subjacente do cliente MaxCompute integrada ao nó MapReduce está sendo atualizada, o que causa falhas nos jobs. Por enquanto, envie jobs usando o cliente MaxCompute.

   • Prepare os dados:

     -- 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');

   • Código de exemplo do 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
     ;

   • Resultado de exemplo: Execute o comando select * from output_table_float_smoke where pt='20230116'; para consultar a tabela de resultados.

     +------------+------------+------------+------------+
     | 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   |
     +------------+------------+------------+------------+