DataWorks：Notebook 進階開發

開發環境調試

1. 掛載資料集：進入個人開發環境詳情頁，在儲存配置 > 資料集中配置資料集。

2. 在代碼中訪問：資料集將被掛載到個人開發環境的掛載路徑下，在代碼中直接讀寫該路徑。

   # 假設已在個人開發環境設定了掛載路徑為/mnt/data/dataset的資料集
   import pandas as pd
   
   # 直接使用掛載路徑
   file_path = '/mnt/data/dataset/testfile.csv'
   df = pd.read_csv(file_path)
   
   # 使用 PyODPS 將資料寫入 MaxCompute
   o = %odps
   o.write_table('mc_test_table', df, overwrite=True)
   print(f"成功將資料寫入 MaxCompute 表 mc_test_table。")

生產環境部署

1. 掛載資料集：在 Notebook 節點編輯頁面的右側導覽列，進入調度配置 > 調度策略，添加上述資料集。

2. 在代碼中訪問：提交發布之後，資料集將被掛載到生產環境的掛載路徑下，在代碼中直接讀寫該路徑。

   # 假設已在個人開發環境設定了掛載路徑為/mnt/data/dataset的資料集
   import pandas as pd
   
   # 直接使用掛載路徑
   file_path = '/mnt/data/dataset/testfile.csv'
   df = pd.read_csv(file_path)
   
   # 使用 PyODPS 將資料寫入 MaxCompute
   o = %odps
   o.write_table('mc_test_table', df, overwrite=True)
   print(f"成功將資料寫入 MaxCompute 表 mc_test_table。")

引擎對比

MaxCompute Spark

通過 Livy 串連到 MaxCompute 專案內建的 Spark 引擎。

1. 建立串連：在 Python 儲存格中運行以下命令。系統將自動建立或複用 Spark 會話。

   # 建立Spark Session，並停止Livy。
   %maxcompute_spark

2. 執行 PySpark 代碼：串連成功後，在新的 Python 儲存格中，使用 %%spark Cell Magic 執行 PySpark 代碼。

   # 在使用MaxCompute Spark時，Python儲存格中必須以%%spark作為首行代碼
   %%spark
   
   df = spark.sql("SELECT * FROM your_mc_table LIMIT 10")
   df.show()

3. 手動釋放串連：在開發調試結束後，可手動停止或刪除會話。在生產環境運行時，系統將自動停止並刪除當前任務執行個體的Livy，無需手動處理。

   # 清理Spark Session，並停止Livy。
   %maxcompute_spark stop
   
   # 清理Spark Session，並停止Livy，最後刪除Livy
   %maxcompute_spark delete

EMR Serverless Spark

通過串連您預先建立的 Livy Gateway 與 EMR Serverless Spark 互動。

1. 建立串連：在執行命令前，需在儲存格右下角選擇要串連的 EMR 計算資源 和 Livy Gateway。

   # 基礎串連
   %emr_serverless_spark
   
   # 或在串連時傳入自訂 Spark 參數，注意需要自訂Spark參數時，需要寫兩個百分比符號
   %%emr_serverless_spark
   {
     "spark_conf": {
       "spark.emr.serverless.environmentId": "<EMR Serverless Spark 運行環境ID>",
       "spark.emr.serverless.network.service.name": "<EMR Serverless Spark 網路連接ID>",
       "spark.driver.cores": "1",
       "spark.driver.memory": "8g",
       "spark.executor.cores": "1",
       "spark.executor.memory": "2g",
       "spark.driver.maxResultSize": "32g"
     }
   }

   說明

   自訂參數與全域配置的關係

   • 預設行為：您在此處定義的自訂參數僅對本次串連（Session）生效，是一次性的。如果您未提供自訂參數，系統將自動使用在管理中心中配置的全域參數。

   • 推薦用法：對於需要在多個任務或被多人複用的配置，建議在管理中心 > Serverless Spark > SPARK參數中進行全域配置，以保證一致性並方便統一管理。

   • 優先順序規則：當同一個參數在自訂參數和全域配置中都被設定時，哪個最終生效，取決於管理中心中的全局配置是否優先選項：

     • 勾選：全域配置將覆蓋本次的自訂參數。

     • 不勾選：本次的自訂參數將覆蓋全域配置。

2. （可選）重新串連：若Livy gateway頁面的token被管理員誤刪後可通過該命令重新建立。

   # 重新串連，重新整理當前個人開發環境的livy token
   %emr_serverless_spark refresh_token

3. 執行 PySpark 或 SQL 代碼：串連成功後，核心已切換。您可以在 Python 儲存格 中直接編寫 PySpark 代碼，或在 EMR Spark SQL 儲存格 中編寫 SQL。

   1. 通過EMR Spark SQL向計算資源提交並執行SQL代碼

      經過%emr_serverless_spark成功建立串連後，可在EMR Spark SQL Cell中，直接寫SQL語句，無需在儲存格中選擇計算資源。

      EMR Spark SQL Cell將複用%emr_serverless_spark的串連，提交至目標計算資源中運行。

      

   2. 通過Python向計算資源提交並執行PySpark代碼

      通過 %emr_serverless_spark 成功建立串連後，可以在新的Python儲存格中提交和執行PySpark代碼，無需在儲存格中添加%%spark首碼。

      

4. 手動釋放串連

   重要

   在多人共用一個 Livy Gateway 的情況下，`stop` 或 `delete` 命令會影響所有正在使用該網關的使用者，請謹慎操作。

   # 清理Spark Session，並停止Livy。
   %emr_serverless_spark stop
   
   # 清理Spark Session，並停止Livy，最後刪除Livy
   %emr_serverless_spark delete

AnalyticDB for Spark

通過建立 Spark Connect Server 串連到 AnalyticDB for Spark 引擎。

1. 建立串連：為確保網路連通，必須在串連參數中正確配置交換器 ID 和安全性群組 ID。 在執行命令前，需在儲存格右下角選擇 ADB Spark 計算資源。

   # 必須配置交換器ID和安全性群組ID以建立網路連接
   %adb_spark add \
    --spark-conf spark.adb.version=3.5 \
    --spark-conf spark.adb.eni.enabled=true \
    --spark-conf spark.adb.eni.vswitchId=<ADB的交換器ID> \
    --spark-conf spark.adb.eni.securityGroupId=<個人開發環境的安全性群組ID>

   如何尋找交換器ID和安全性群組ID？

   • 交換器ID (vswitchId)：前往阿里雲AnalyticDB MySQL控制台，在執行個體詳情頁查看網路資訊的交換器ID。

   • 安全性群組ID (securityGroupId)：進入個人開發環境詳情頁的網路設定中所選安全性群組，查看以sg-開頭的安全性群組ID。

     重要

     為確保網路連通，建立個人開發環境時，建議選擇與您的AnalyticDB for Spark執行個體相同的VPC和交換器。

2. 執行 PySpark 代碼：串連成功後，在新的 Python 儲存格中，執行 PySpark 代碼。

   # 只能對 C-Store 表執行操作
   df = spark.sql("SELECT * FROM my_adb_cstore_table LIMIT 10")
   df.show()

   說明：AnalyticDB for Spark 引擎當前只能處理具有 'storagePolicy'='COLD' 屬性的 C-Store 表。

3. 手動釋放串連：在開發環境調試結束後，手動清理串連會話以節約資源。生產環境運行時，系統將自動清理資源。

   %adb_spark cleanup