MaxCompute：利用MaxCompute External Volume處理非結構化資料

利用Spark on MaxCompute通過External Volume引用或處理OSS資料

Spark on MaxCompute是MaxCompute提供的相容開源Spark的計算服務。它在統一的計算資源和資料集許可權體系之上，提供Spark計算架構，支援您以熟悉的開發使用方式提交運行Spark作業，滿足更豐富的資料處理分析需求。Spark在運行過程中需要載入作業運行資源（File、Archive），其中一種方式是使用Spark直接存取OSS，詳情請參見Spark訪問OSS。如果需要對資源和資料做細粒度的許可權控制，則使用External Volume的方式，通過數倉的許可權體系，對資源做存取控制。

-- 配置項
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

-- 代碼
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()

利用Proxima CE在MaxCompute中做向量計算

使用Proxima CE在MaxCompute中做向量計算，使用說明和樣本如下：

1. 安裝Proxima CE資源套件。

2. 運行任務。

   • 使用限制：

     • Proxima Java SDK目前只支援在Linux和Mac系統下的MaxCompute用戶端執行任務命令。

       說明

       Proxima CE在運行時分為兩部分：本地運行任務和MaxCompute任務。本地運行任務是指裡面沒有涉及到MaxCompute的SQL、MapReduce和Graph任務的功能模組部分；MaxCompute任務是指基於MaxCompute的SQL、MapReduce和Graph等引擎執行的任務，二者會交替執行。Proxima CE運行後，會先嘗試在本地機器上（使用MaxCompute用戶端運行Proxima CE的機器）載入Proxima核心，如果成功則會在本地運行某些模組調用基於 Proxima 核心的函數；如果載入失敗會報錯，但不影響後續運行，模組調用其他函數替代。由於任務jar包內部為Linux相關依賴，因此不支援在Windowss系統下的MaxCompute用戶端運行。

     • 暫不支援通過DataWorks的MapReduce節點執行任務。因為MapReduce節點整合的底層MaxCompute用戶端版本正在升級中，任務會執行失敗，請您暫時用MaxCompute用戶端提交任務。

   • 資料準備：

     -- 建立輸入表
     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);
     
     -- 插入doc資料（底庫表）
     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');
     
     -- 插入query資料（查詢表）
     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');

   • 樣本任務代碼：

     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
     ;

   • 樣本結果：使用select * from output_table_float_smoke where pt='20230116';命令查詢結果表。

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