MaxCompute：Gunakan volume eksternal MaxCompute untuk memproses data tak terstruktur

Gunakan Spark on MaxCompute untuk merujuk atau memproses data OSS berdasarkan volume eksternal

Spark on MaxCompute adalah layanan komputasi yang disediakan oleh MaxCompute dan kompatibel dengan Spark sumber terbuka. Spark on MaxCompute menyediakan kerangka kerja komputasi Spark berdasarkan integrasi sumber daya komputasi, dataset, dan sistem izin. Layanan ini memungkinkan Anda menggunakan metode pengembangan pilihan Anda untuk mengirimkan dan menjalankan pekerjaan Spark. Spark on MaxCompute dapat memenuhi berbagai kebutuhan pemrosesan dan analitik data. Saat menjalankan pekerjaan Spark, sumber daya terkait seperti file dan arsip perlu dimuat. Anda dapat langsung mengakses OSS dari Spark on MaxCompute untuk memuat sumber daya terkait. Untuk informasi lebih lanjut, lihat Akses OSS dari Spark on MaxCompute. Jika Anda perlu melakukan kontrol izin mendetail pada sumber daya dan data, Anda dapat menggunakan volume eksternal untuk melakukan kontrol akses sumber daya berdasarkan sistem izin MaxCompute.

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

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

    # Muat dan analisis 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(' ')]))

    # Bangun model (kelompokkan data)
    clusters = KMeans.train(parsedData, 2, maxIterations=10, initializationMode="random")

    # Evaluasi pengelompokan dengan menghitung 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))

    # Simpan dan muat 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()

Gunakan Proxima CE untuk melakukan vektorisasi di MaxCompute

Bagian ini memberikan instruksi dan contoh untuk menggunakan Proxima CE melakukan vektorisasi di MaxCompute.

1. Instal paket sumber daya Proxima CE.

2. Jalankan tugas.

   • Batasan:

     • SDK Proxima untuk Java memungkinkan Anda menjalankan perintah tugas pada klien MaxCompute yang berjalan di sistem operasi Linux atau macOS.

       Catatan

       Saat menjalankan Proxima CE, dua jenis tugas terlibat: tugas lokal dan tugas MaxCompute. Tugas lokal adalah tugas yang tidak melibatkan SQL, MapReduce, dan tugas Graph di MaxCompute. Tugas MaxCompute adalah tugas yang dieksekusi berdasarkan mesin komputasi MaxCompute seperti SQL, MapReduce, dan Graph. Kedua jenis tugas tersebut dapat dijalankan secara bergantian. Setelah Anda menjalankan Proxima CE, ia pertama-tama mencoba memuat kernel Proxima pada mesin lokal tempat Proxima CE dijalankan menggunakan klien MaxCompute. Jika kernel Proxima berhasil dimuat, modul tertentu dijalankan pada mesin lokal untuk memanggil fungsi-fungsi berbasis kernel Proxima. Jika operasi pemuatan gagal, kesalahan dilaporkan. Namun, operasi selanjutnya tidak terpengaruh secara negatif, dan modul-modul tersebut memanggil fungsi lain sebagai gantinya. Paket JAR berisi dependensi terkait Linux. Oleh karena itu, paket JAR tidak dapat dijalankan pada klien MaxCompute di sistem operasi Windows.

     • Anda tidak dapat menggunakan node MapReduce DataWorks untuk mengeksekusi tugas. Hal ini karena versi klien MaxCompute bawah yang terintegrasi dengan node MapReduce sedang ditingkatkan, dan tugas gagal dieksekusi. Kami sarankan Anda menggunakan klien MaxCompute untuk mengirimkan tugas.

   • Persiapan data:

     -- Buat tabel input.
     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);
     
     -- Masukkan data ke dalam tabel doc_table_float_smoke (tabel dasar).
     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');
     
     -- Masukkan data ke dalam tabel query_table_float_smoke (tabel kueri).
     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');

   • Kode tugas contoh:

     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
     ;

   • Hasil contoh: Jalankan pernyataan select * from output_table_float_smoke where pt='20230116'; untuk menanyakan data dari tabel hasil.

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