關(guān)注TechLead，分享AI全維度知識(shí)。作者擁有10+年互聯(lián)網(wǎng)服務(wù)架構(gòu)、AI產(chǎn)品研發(fā)經(jīng)驗(yàn)、團(tuán)隊(duì)管理經(jīng)驗(yàn)，同濟(jì)本復(fù)旦碩，復(fù)旦機(jī)器人智能實(shí)驗(yàn)室成員，阿里云認(rèn)證的資深架構(gòu)師，項(xiàng)目管理專業(yè)人士，上億營(yíng)收AI產(chǎn)品研發(fā)負(fù)責(zé)人。文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-714117.html

本文內(nèi)容

數(shù)據(jù)整理已經(jīng)成為機(jī)器學(xué)習(xí)項(xiàng)目中最重要的步驟之一。 Azure 機(jī)器學(xué)習(xí)與 Azure Synapse Analytics 集成，提供對(duì) Apache Spark Pool（由 Azure Synapse 支持）的訪問(wèn)，以便使用 Azure 機(jī)器學(xué)習(xí)筆記本進(jìn)行交互式數(shù)據(jù)整理。

先決條件

• 一個(gè) Azure 訂閱；如果你沒(méi)有 Azure 訂閱，請(qǐng)?jiān)陂_(kāi)始之前創(chuàng)建一個(gè)免費(fèi)帳戶。
• Azure 機(jī)器學(xué)習(xí)工作區(qū)。 請(qǐng)參閱創(chuàng)建工作區(qū)資源。
• Azure Data Lake Storage (ADLS) Gen 2 存儲(chǔ)帳戶。 請(qǐng)參閱創(chuàng)建 Azure Data Lake Storage (ADLS) Gen 2 存儲(chǔ)帳戶。
• （可選）：Azure Key Vault。 請(qǐng)參閱創(chuàng)建 Azure 密鑰保管庫(kù)。
• （可選）：服務(wù)主體。 請(qǐng)參閱創(chuàng)建服務(wù)主體。
• （可選）Azure 機(jī)器學(xué)習(xí)工作區(qū)中附加的 Synapse Spark 池。

在開(kāi)始數(shù)據(jù)整理任務(wù)之前，請(qǐng)了解存儲(chǔ)機(jī)密的過(guò)程

• Azure Blob 存儲(chǔ)帳戶訪問(wèn)密鑰
• 共享訪問(wèn)簽名 (SAS) 令牌
• Azure Data Lake Storage (ADLS) Gen 2 服務(wù)主體信息

在 Azure 密鑰保管庫(kù)中。 還需要了解如何在 Azure 存儲(chǔ)帳戶中處理角色分配。 下面的部分討論以下概念。 然后，我們將詳細(xì)了解如何使用 Azure 機(jī)器學(xué)習(xí)筆記本中的 Spark 池進(jìn)行交互式數(shù)據(jù)整理。

使用 Apache Spark 進(jìn)行交互式數(shù)據(jù)整理

Azure 機(jī)器學(xué)習(xí)在 Azure 機(jī)器學(xué)習(xí)筆記本中提供無(wú)服務(wù)器 Spark 計(jì)算和附加的 Synapse Spark 池，用于與 Apache Spark 進(jìn)行交互式數(shù)據(jù)整理。 無(wú)服務(wù)器 Spark 計(jì)算不需要在 Azure Synapse 工作區(qū)中創(chuàng)建資源。 相反，在 Azure 機(jī)器學(xué)習(xí)筆記本中可以直接使用完全托管的無(wú)服務(wù)器 Spark 計(jì)算。 要訪問(wèn) Azure 機(jī)器學(xué)習(xí)中的 Spark 群集，最簡(jiǎn)單的方法是使用無(wú)服務(wù)器 Spark 計(jì)算。

Azure 機(jī)器學(xué)習(xí)筆記本中的無(wú)服務(wù)器 Spark 計(jì)算

默認(rèn)情況下，Azure 機(jī)器學(xué)習(xí)筆記本中提供了無(wú)服務(wù)器 Spark 計(jì)算。 若要在筆記本中訪問(wèn)它，請(qǐng)從“計(jì)算”選擇菜單的“Azure 機(jī)器學(xué)習(xí)無(wú)服務(wù)器 Spark”下選擇“無(wú)服務(wù)器 Spark 計(jì)算”。

筆記本 UI 還為無(wú)服務(wù)器 Spark 計(jì)算提供了 Spark 會(huì)話配置選項(xiàng)。 配置 Spark 會(huì)話：

1. 選擇屏幕頂部的“配置會(huì)話”。

2. 從下拉菜單中選擇“Apache Spark 版本”。

   重要

   適用于 Apache Spark 的 Azure Synapse 運(yùn)行時(shí)：公告

   • 適用于 Apache Spark 3.2 的 Azure Synapse 運(yùn)行時(shí)：
     • EOLA 公告日期：2023 年 7 月 8 日
     • 支持結(jié)束日期：2024 年 7 月 8 日。 在此日期之后，將會(huì)禁用運(yùn)行時(shí)。
   • 為了獲取持續(xù)支持和最佳性能，建議遷移到 Apache Sark 3.3。

3. 從下拉菜單中選擇“實(shí)例類型”。 當(dāng)前支持以下實(shí)例類型：

   • Standard_E4s_v3
   • Standard_E8s_v3
   • Standard_E16s_v3
   • Standard_E32s_v3
   • Standard_E64s_v3

4. 輸入 Spark 會(huì)話超時(shí)值（以分鐘為單位）。

5. 選擇是否動(dòng)態(tài)分配執(zhí)行程序

6. 選擇 Spark 會(huì)話的執(zhí)行程序數(shù)量。

7. 從下拉菜單中選擇“執(zhí)行程序大小”。

8. 從下拉菜單中選擇“驅(qū)動(dòng)程序大小”。

9. 要使用 Conda 文件配置 Spark 會(huì)話，請(qǐng)選中“上傳 conda 文件”復(fù)選框。 然后，選擇“瀏覽”，并選擇具有所需 Spark 會(huì)話配置的 Conda 文件。

10. 添加“配置設(shè)置”屬性，在“屬性”和“值”文本框中輸入值，然后選擇“添加”。

11. 選擇“應(yīng)用”。

12. 在“配置新會(huì)話?”彈出窗口中選擇“停止會(huì)話”。

會(huì)話配置更改將被保存，并可用于使用無(wú)服務(wù)器 Spark 計(jì)算啟動(dòng)的另一個(gè)筆記本會(huì)話。

提示

如果使用會(huì)話級(jí) Conda 包，并將配置變量 spark.hadoop.aml.enable_cache 設(shè)置為 true，則可以改善 Spark 會(huì)話冷啟動(dòng)時(shí)間。 會(huì)話首次啟動(dòng)時(shí)，具有會(huì)話級(jí)別 Conda 包的會(huì)話冷啟動(dòng)通常需要 10 到 15 分鐘。 但是，配置變量設(shè)置為 true 時(shí)的后續(xù)會(huì)話冷啟動(dòng)通常需要 3 到 5 分鐘。

從 Azure Data Lake Storage (ADLS) Gen 2 導(dǎo)入和整理數(shù)據(jù)

可以使用 abfss:// 數(shù)據(jù) URI 按照以下兩種數(shù)據(jù)訪問(wèn)機(jī)制之一訪問(wèn)和處理存儲(chǔ)在 Azure Data Lake Storage (ADLS) Gen 2 存儲(chǔ)帳戶中的數(shù)據(jù)：

• 用戶標(biāo)識(shí)傳遞
• 基于服務(wù)主體的數(shù)據(jù)訪問(wèn)

提示

要使用無(wú)服務(wù)器 Spark 計(jì)算進(jìn)行數(shù)據(jù)整理、對(duì) Azure Data Lake Storage (ADLS) Gen 2 存儲(chǔ)帳戶中的數(shù)據(jù)進(jìn)行用戶標(biāo)識(shí)直通訪問(wèn)，需要的配置步驟是最少的。

若要使用用戶標(biāo)識(shí)傳遞開(kāi)始交互式數(shù)據(jù)整理，請(qǐng)執(zhí)行以下命令：

• 驗(yàn)證用戶身份是否在 Azure Data Lake Storage (ADLS) Gen 2 存儲(chǔ)帳戶中獲得了“參與者”和“存儲(chǔ) Blob 數(shù)據(jù)參與者”角色。

• 要使用無(wú)服務(wù)器 Spark 計(jì)算，請(qǐng)?jiān)凇坝?jì)算”選擇菜單中，選擇“Azure 機(jī)器學(xué)習(xí)無(wú)服務(wù)器 Spark”下的“無(wú)服務(wù)器 Spark 計(jì)算”。

• 要使用附加的 Synapse Spark 池，請(qǐng)從“計(jì)算”選擇菜單中選擇“Synapse Spark 池”下附加的 Synapse Spark 池。

• 這個(gè) Titanic 數(shù)據(jù)整理代碼示例顯示了 abfss://<FILE_SYSTEM_NAME>@<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net/<PATH_TO_DATA> 格式的數(shù)據(jù) URI 與 pyspark.pandas 和 pyspark.ml.feature.Imputer 的搭配使用。

  import pyspark.pandas as pd
  from pyspark.ml.feature import Imputer
  
  df = pd.read_csv(
      "abfss://<FILE_SYSTEM_NAME>@<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net/data/titanic.csv",
      index_col="PassengerId",
  )
  imputer = Imputer(inputCols=["Age"], outputCol="Age").setStrategy(
      "mean"
  )  # Replace missing values in Age column with the mean value
  df.fillna(
      value={"Cabin": "None"}, inplace=True
  )  # Fill Cabin column with value "None" if missing
  df.dropna(inplace=True)  # Drop the rows which still have any missing value
  df.to_csv(
      "abfss://<FILE_SYSTEM_NAME>@<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net/data/wrangled",
      index_col="PassengerId",
  )
  

  備注

  此 Python 代碼示例使用 pyspark.pandas。 只有 Spark 運(yùn)行時(shí)版本 3.2 或更高版本才支持此功能。

若要通過(guò)服務(wù)主體按照訪問(wèn)權(quán)限來(lái)整理數(shù)據(jù)，請(qǐng)執(zhí)行以下操作：

1. 驗(yàn)證服務(wù)主體是否在 Azure Data Lake Storage (ADLS) Gen 2 存儲(chǔ)帳戶中獲得了“參與者”和“存儲(chǔ) Blob 數(shù)據(jù)參與者”角色。

2. 為服務(wù)主體租戶 ID、客戶端 ID 和客戶端機(jī)密值創(chuàng)建 Azure 密鑰保管庫(kù)機(jī)密。

3. 在“計(jì)算”選擇菜單中，選擇“Azure 機(jī)器學(xué)習(xí)無(wú)服務(wù)器 Spark”下的“無(wú)服務(wù)器 Spark 計(jì)算”，或者從“計(jì)算”選擇菜單中選擇“Synapse Spark 池”下附加的 Synapse Spark 池。

4. 若要在配置中設(shè)置服務(wù)主體租戶 ID、客戶端 ID 和客戶端密碼，請(qǐng)執(zhí)行以下代碼示例。

   • 代碼中的 get_secret() 調(diào)用取決于 Azure 密鑰保管庫(kù)的名稱，以及為服務(wù)主體租戶 ID、客戶端 ID 和客戶端密碼創(chuàng)建的 Azure 密鑰保管庫(kù)機(jī)密的名稱。 在配置中設(shè)置這些相應(yīng)的屬性名稱/值：

     • 客戶端 ID 屬性：fs.azure.account.oauth2.client.id.<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net
     • 客戶端機(jī)密屬性：fs.azure.account.oauth2.client.secret.<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net
     • 租戶 ID 屬性：fs.azure.account.oauth2.client.endpoint.<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net
     • 租戶 ID 值：https://login.microsoftonline.com/<TENANT_ID>/oauth2/token

     from pyspark.sql import SparkSession
     
     sc = SparkSession.builder.getOrCreate()
     token_library = sc._jvm.com.microsoft.azure.synapse.tokenlibrary.TokenLibrary
     
     # Set up service principal tenant ID, client ID and secret from Azure Key Vault
     client_id = token_library.getSecret("<KEY_VAULT_NAME>", "<CLIENT_ID_SECRET_NAME>")
     tenant_id = token_library.getSecret("<KEY_VAULT_NAME>", "<TENANT_ID_SECRET_NAME>")
     client_secret = token_library.getSecret("<KEY_VAULT_NAME>", "<CLIENT_SECRET_NAME>")
     
     # Set up service principal which has access of the data
     sc._jsc.hadoopConfiguration().set(
         "fs.azure.account.auth.type.<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net", "OAuth"
     )
     sc._jsc.hadoopConfiguration().set(
         "fs.azure.account.oauth.provider.type.<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net",
         "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider",
     )
     sc._jsc.hadoopConfiguration().set(
         "fs.azure.account.oauth2.client.id.<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net",
         client_id,
     )
     sc._jsc.hadoopConfiguration().set(
         "fs.azure.account.oauth2.client.secret.<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net",
         client_secret,
     )
     sc._jsc.hadoopConfiguration().set(
         "fs.azure.account.oauth2.client.endpoint.<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net",
         "https://login.microsoftonline.com/" + tenant_id + "/oauth2/token",
     )
     

5. 使用 abfss://<FILE_SYSTEM_NAME>@<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net/<PATH_TO_DATA> 格式的數(shù)據(jù) URI 導(dǎo)入和轉(zhuǎn)換數(shù)據(jù)，如使用 Titanic 數(shù)據(jù)的代碼示例所示。
   
   

從 Azure Blob 存儲(chǔ)導(dǎo)入和處理數(shù)據(jù)

可以使用存儲(chǔ)帳戶訪問(wèn)密鑰或共享訪問(wèn)簽名 (SAS) 令牌訪問(wèn) Azure Blob 存儲(chǔ)數(shù)據(jù)。 應(yīng)將這些憑據(jù)作為機(jī)密存儲(chǔ)在 Azure 密鑰保管庫(kù)中，并在會(huì)話配置中將其設(shè)置為屬性。

若要開(kāi)始交互式數(shù)據(jù)整理，請(qǐng)執(zhí)行以下操作：

1. 在左側(cè)Azure 機(jī)器學(xué)習(xí)工作室面板中，選擇“筆記本”。

2. 在“計(jì)算”選擇菜單中，選擇“Azure 機(jī)器學(xué)習(xí)無(wú)服務(wù)器 Spark”下的“無(wú)服務(wù)器 Spark 計(jì)算”，或者從“計(jì)算”選擇菜單中選擇“Synapse Spark 池”下附加的 Synapse Spark 池。

3. 若要配置存儲(chǔ)帳戶訪問(wèn)密鑰或共享訪問(wèn)簽名 (SAS) 令牌以便在 Azure 機(jī)器學(xué)習(xí)筆記本中訪問(wèn)數(shù)據(jù)，請(qǐng)執(zhí)行以下操作：

   • 對(duì)于訪問(wèn)密鑰，請(qǐng)?jiān)O(shè)置屬性 fs.azure.account.key.<STORAGE_ACCOUNT_NAME>.blob.core.windows.net，如以下代碼片段所示：

     from pyspark.sql import SparkSession
     
     sc = SparkSession.builder.getOrCreate()
     token_library = sc._jvm.com.microsoft.azure.synapse.tokenlibrary.TokenLibrary
     access_key = token_library.getSecret("<KEY_VAULT_NAME>", "<ACCESS_KEY_SECRET_NAME>")
     sc._jsc.hadoopConfiguration().set(
         "fs.azure.account.key.<STORAGE_ACCOUNT_NAME>.blob.core.windows.net", access_key
     )
     

   • 對(duì)于 SAS 令牌，請(qǐng)?jiān)O(shè)置屬性 fs.azure.sas.<BLOB_CONTAINER_NAME>.<STORAGE_ACCOUNT_NAME>.blob.core.windows.net，如以下代碼片段所示：

     from pyspark.sql import SparkSession
     
     sc = SparkSession.builder.getOrCreate()
     token_library = sc._jvm.com.microsoft.azure.synapse.tokenlibrary.TokenLibrary
     sas_token = token_library.getSecret("<KEY_VAULT_NAME>", "<SAS_TOKEN_SECRET_NAME>")
     sc._jsc.hadoopConfiguration().set(
         "fs.azure.sas.<BLOB_CONTAINER_NAME>.<STORAGE_ACCOUNT_NAME>.blob.core.windows.net",
         sas_token,
     )
     

     備注

     上述代碼片段中的 get_secret() 調(diào)用需要 Azure 密鑰庫(kù)的名稱，以及為 Azure Blob 存儲(chǔ)帳戶訪問(wèn)密鑰或 SAS 令牌創(chuàng)建的機(jī)密的名稱

4. 在同一筆記本中執(zhí)行數(shù)據(jù)整理代碼。 將數(shù)據(jù) URI 的格式設(shè)置為 wasbs://<BLOB_CONTAINER_NAME>@<STORAGE_ACCOUNT_NAME>.blob.core.windows.net/<PATH_TO_DATA>，類似于此代碼片段所示：

   import pyspark.pandas as pd
   from pyspark.ml.feature import Imputer
   
   df = pd.read_csv(
       "wasbs://<BLOB_CONTAINER_NAME>@<STORAGE_ACCOUNT_NAME>.blob.core.windows.net/data/titanic.csv",
       index_col="PassengerId",
   )
   imputer = Imputer(inputCols=["Age"], outputCol="Age").setStrategy(
       "mean"
   )  # Replace missing values in Age column with the mean value
   df.fillna(
       value={"Cabin": "None"}, inplace=True
   )  # Fill Cabin column with value "None" if missing
   df.dropna(inplace=True)  # Drop the rows which still have any missing value
   df.to_csv(
       "wasbs://<BLOB_CONTAINER_NAME>@<STORAGE_ACCOUNT_NAME>.blob.core.windows.net/data/wrangled",
       index_col="PassengerId",
   )
   

   備注

   此 Python 代碼示例使用 pyspark.pandas。 只有 Spark 運(yùn)行時(shí)版本 3.2 或更高版本才支持此功能。

從 Azure 機(jī)器學(xué)習(xí)數(shù)據(jù)存儲(chǔ)導(dǎo)入和整理數(shù)據(jù)

若要從 Azure 機(jī)器學(xué)習(xí)數(shù)據(jù)存儲(chǔ)訪問(wèn)數(shù)據(jù)，請(qǐng)使用 URI 格式azureml://datastores/<DATASTORE_NAME>/paths/<PATH_TO_DATA>定義數(shù)據(jù)存儲(chǔ)上數(shù)據(jù)的路徑。 若要在筆記本會(huì)話中以交互方式處理 Azure 機(jī)器學(xué)習(xí)數(shù)據(jù)存儲(chǔ)中的數(shù)據(jù)，請(qǐng)執(zhí)行以下操作：

1. 在“計(jì)算”選擇菜單中，選擇“Azure 機(jī)器學(xué)習(xí)無(wú)服務(wù)器 Spark”下的“無(wú)服務(wù)器 Spark 計(jì)算”，或者從“計(jì)算”選擇菜單中選擇“Synapse Spark 池”下附加的 Synapse Spark 池。

2. 此代碼示例顯示了如何使用 azureml:// 數(shù)據(jù)存儲(chǔ) URI pyspark.pandas 和 pyspark.ml.feature.Imputer 從 Azure 機(jī)器學(xué)習(xí)數(shù)據(jù)存儲(chǔ)中讀取和處理大量數(shù)據(jù)。

   import pyspark.pandas as pd
   from pyspark.ml.feature import Imputer
   
   df = pd.read_csv(
       "azureml://datastores/workspaceblobstore/paths/data/titanic.csv",
       index_col="PassengerId",
   )
   imputer = Imputer(inputCols=["Age"], outputCol="Age").setStrategy(
       "mean"
   )  # Replace missing values in Age column with the mean value
   df.fillna(
       value={"Cabin": "None"}, inplace=True
   )  # Fill Cabin column with value "None" if missing
   df.dropna(inplace=True)  # Drop the rows which still have any missing value
   df.to_csv(
       "azureml://datastores/workspaceblobstore/paths/data/wrangled",
       index_col="PassengerId",
   )
   

   備注

   此 Python 代碼示例使用 pyspark.pandas。 只有 Spark 運(yùn)行時(shí)版本 3.2 或更高版本才支持此功能。

Azure 機(jī)器學(xué)習(xí)數(shù)據(jù)存儲(chǔ)可以使用 Azure 存儲(chǔ)帳戶憑據(jù)訪問(wèn)數(shù)據(jù)

• 訪問(wèn)密鑰
• SAS 令牌
• 服務(wù)主體 (service principal)

或提供無(wú)憑據(jù)的數(shù)據(jù)訪問(wèn)。 根據(jù)數(shù)據(jù)存儲(chǔ)類型和基礎(chǔ) Azure 存儲(chǔ)帳戶類型，選擇適當(dāng)?shù)纳矸蒡?yàn)證機(jī)制來(lái)確保數(shù)據(jù)訪問(wèn)。 下表總結(jié)了用于訪問(wèn) Azure 機(jī)器學(xué)習(xí)數(shù)據(jù)存儲(chǔ)中的數(shù)據(jù)的身份驗(yàn)證機(jī)制：

只有在未啟用軟刪除的情況下，_* 用戶標(biāo)識(shí)直通才適用于指向 Azure Blob 存儲(chǔ)帳戶的無(wú)憑據(jù)數(shù)據(jù)存儲(chǔ)。

默認(rèn)文件共享掛載到無(wú)服務(wù)器 Spark 計(jì)算和附加的 Synapse Spark 池。

在 Azure 機(jī)器學(xué)習(xí)工作室中，默認(rèn)文件共享中的文件顯示在“文件”選項(xiàng)卡下的目錄樹中。筆記本代碼可以使用 file:// 協(xié)議以及文件的絕對(duì)路徑直接訪問(wèn)此文件共享中存儲(chǔ)的文件，而無(wú)需進(jìn)行更多配置。 此代碼片段演示如何訪問(wèn)存儲(chǔ)在默認(rèn)文件共享上的文件：

import os
import pyspark.pandas as pd
from pyspark.ml.feature import Imputer

abspath = os.path.abspath(".")
file = "file://" + abspath + "/Users/<USER>/data/titanic.csv"
print(file)
df = pd.read_csv(file, index_col="PassengerId")
imputer = Imputer(
    inputCols=["Age"],
    outputCol="Age").setStrategy("mean") # Replace missing values in Age column with the mean value
df.fillna(value={"Cabin" : "None"}, inplace=True) # Fill Cabin column with value "None" if missing
df.dropna(inplace=True) # Drop the rows which still have any missing value
output_path = "file://" + abspath + "/Users/<USER>/data/wrangled"
df.to_csv(output_path, index_col="PassengerId")

備注

此 Python 代碼示例使用 pyspark.pandas。 只有 Spark 運(yùn)行時(shí)版本 3.2 或更高版本才支持此功能。

關(guān)注TechLead，分享AI全維度知識(shí)。作者擁有10+年互聯(lián)網(wǎng)服務(wù)架構(gòu)、AI產(chǎn)品研發(fā)經(jīng)驗(yàn)、團(tuán)隊(duì)管理經(jīng)驗(yàn)，同濟(jì)本復(fù)旦碩，復(fù)旦機(jī)器人智能實(shí)驗(yàn)室成員，阿里云認(rèn)證的資深架構(gòu)師，項(xiàng)目管理專業(yè)人士，上億營(yíng)收AI產(chǎn)品研發(fā)負(fù)責(zé)人。

到了這里，關(guān)于Azure - 機(jī)器學(xué)習(xí)：使用 Apache Spark 進(jìn)行交互式數(shù)據(jù)整理的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！