如何使用Notebook開發Spark作業 - AnalyticDB

Notebook開發是一個互動式資料分析開發平台，提供了作業編輯、資料分析、資料視覺效果等功能。如果您需要使用Notebook開發Spark SQL作業，可以參考本文檔，使用DMS的Notebook功能完成作業開發。

前提條件

AnalyticDB for MySQL叢集的產品系列為企業版、基礎版或湖倉版。
已在AnalyticDB for MySQL叢集中建立Job型資源群組。
已建立AnalyticDB for MySQL叢集的資料庫帳號。
- 如果是通過阿里雲帳號訪問，只需建立高許可權帳號。
- 如果是通過RAM使用者訪問，需要建立高許可權帳號和普通帳號並且將RAM使用者綁定到普通帳號上。
已授權AnalyticDB for MySQL扮演AliyunADBSparkProcessingDataRole角色來訪問其他雲資源。
AnalyticDB for MySQL叢集已配置Spark應用的日誌儲存地址。
說明
登入雲原生資料倉儲AnalyticDB MySQL控制台，在作業開發 > Spark Jar 開發頁面，單擊日誌配置，選擇預設路徑或自訂儲存路徑。自訂儲存路徑時不能將日誌儲存在OSS的根目錄下，請確保該路徑中至少包含一層檔案夾。
AnalyticDB for MySQL叢集與OSS儲存空間位於相同地區。

使用限制

目前僅部分地區支援Notebook功能，包括：華東1（杭州）地區。

操作步驟

建立專案並進入專案空間。
1. 登入Data Management 5.0。
2. 單擊控制台左上方的表徵圖，選擇全部功能 > Data+AI > Notebook。
  說明
  若您使用的是非極簡模式的控制台，請在頂部功能表列中，選擇Data+AI > Notebook。
3. 單擊建立專案空間，在對話方塊中填入專案空間名、地區等資訊，單擊確認。
4. 單擊操作列的進入空間，進入目標專案空間。
（可選）添加專案成員。如果專案空間有多個使用者參與，則需要完成此步驟，給使用者授予不同的專案空間角色。
配置代碼儲存空間。
1. 在專案設定頁簽內，單擊儲存配置。
2. 配置代碼儲存的OSS路徑。

添加資源。

在專案設定頁簽內，單擊資源配置。

單擊添加資源，配置資源的相關資訊。

參數	是否必填	說明
資源名稱	是	資源名稱。您可以自訂。
資源簡介	是	填寫該資源的使用用途。您可以自訂。
鏡像	是	僅支援選擇Spark3.5+Python3.9鏡像。
ADB執行個體	是	AnalyticDB for MySQL叢集ID。說明如果搜尋不到目的地組群，請檢查叢集是否錄入至DMS。
ADB資源群組	是	選擇目標Job型資源群組。
Executor規格	是	選擇Spark Executor的資源規格。本文以預設值medium規格為例。不同型號的取值對應不同的規格，詳情請參見Spark應用配置參數說明的型號列。
Executor上限 Executor下限	是	Spark Executor數量。選擇Spark3.5+Python3.9鏡像後，Executor下限預設為2，Executor上限預設為8。
Notebook規格	是	選擇Notebook規格，本文以General_Tiny_v1（1核4 GB）為例。
VPC ID	是	選擇AnalyticDB for MySQL所屬的VPC，以保證Notebook和AnalyticDB for MySQL可以正常通訊。重要如果後續AnalyticDB for MySQL切換了新的專用網路和交換器，則需要將資源中的VPC ID和VSwitch ID修改為新的專用網路和交換器，否則作業會提交失敗。
Zone ID	是	選擇AnalyticDB for MySQL所在的可用性區域。
VSwitch ID	是	選擇AnalyticDB for MySQL所屬的交換器。
安全性群組ID	是	選擇可用的安全性群組，以保證Notebook和AnalyticDB for MySQL可以正常通訊。
資源釋放	是	資源的閑置狀態到達指定時間長度後，將自動釋放。
依賴的Jars	否	Jar包的OSS儲存路徑。僅通過Python提交作業且使用了Jar包時，填寫該參數。
SparkConf	否	與開源Spark中的配置項基本一致，參數格式為`key: value`形式。與開源Spark用法不一致的配置參數及AnalyticDB for MySQL特有的配置參數，請參見Spark應用配置參數說明。

單擊儲存。
單擊目標資源右側操作列的啟動，啟動資源。

初始化資料。

單擊控制台左上方的表徵圖，選擇全部功能 > 資料資產 > 執行個體管理。

單擊+新增，在彈出的新增執行個體對話方塊中配置如下參數：

參數		說明
資料來源		在阿里雲頁簽下，選擇OSS。
基本資料	檔案及日誌	固定選擇OSS。
	執行個體地區	選擇AnalyticDB for MySQL叢集所在地區。
	錄入方式	固定選擇串連串地址。
	串連串地址	固定填寫oss-cn-hangzhou.aliyuncs.com。
	Bucket	選擇Bucket名稱。
	訪問方式	訪問方式，本文以安全託管-手動為例。
	AccessKey ID	阿里雲帳號或具備OSS存取權限的RAM使用者的AccessKey ID。如何擷取AccessKey ID和AccessKey Secret，請參見帳號與許可權。
	AccessKey Sercert	阿里雲帳號或具備OSS存取權限的RAM使用者的AccessKey Secret。如何擷取AccessKey ID和AccessKey Secret，請參見帳號與許可權。
進階資訊		選填參數。參數說明，請參見進階資訊。

填寫完成以上資訊後，單擊左下角的測試連接。
說明
如果測試連接失敗，請按照報錯提示檢查您錄入的執行個體資訊。
出現串連成功提示後，單擊提交。
進入專案空間，單擊頁簽下。
在湖倉資料頁簽下，單擊添加OSS，選擇步驟b新增的Bucket。

建立Notebook。
在檔案頁簽下，單擊，選擇Notebook。

在Notebook頁面開發Spark SQL作業。

說明

Notebook頁面中各個按鈕的詳細介紹，請參見Notebook介面介紹。

執行以下命令，下載Python依賴：
```
pip install delta
```
將Cell類型切換為SQL，執行以下語句建立資料庫。
說明
步驟b建立的db_delta庫和步驟c建立的sample_data外表會在AnalyticDB for MySQL中自動顯示，後續您可在AnalyticDB for MySQL控制台上分析sample_data表。
```
CREATE DATABASE db_delta 
LOCATION 'oss://testBucketName/db_delta/';    -- 指定db_delta庫中資料的儲存路徑
```

將Cell類型切換為Code，執行以下代碼，建立sample_data外表並插入資料。sample_data外表資料會儲存在步驟b指定的OSS路徑中。

# -*- coding: utf-8 -*-

import pyspark
from delta import *
from pyspark.sql.types import *
from pyspark.sql.functions import *


print("Starting Delta table creation")

data = [
    ("Robert", "Baratheon", "Baratheon", "Storms End", 48),
    ("Eddard", "Stark", "Stark", "Winterfell", 46),
    ("Jamie", "Lannister", "Lannister", "Casterly Rock", 29),
    ("Robert", "Baratheon", "Baratheon", "Storms End", 48),
    ("Eddard", "Stark", "Stark", "Winterfell", 46),
    ("Jamie", "Lannister", "Lannister", "Casterly Rock", 29),
    ("Robert", "Baratheon", "Baratheon", "Storms End", 48),
    ("Eddard", "Stark", "Stark", "Winterfell", 46),
    ("Jamie", "Lannister", "Lannister", "Casterly Rock", 29)
        ]

schema = StructType([
    StructField("firstname", StringType(), True),
    StructField("lastname", StringType(), True),
    StructField("house", StringType(), True),
    StructField("location", StringType(), True),
    StructField("age", IntegerType(), True)
])

sample_dataframe = spark.createDataFrame(data=data, schema=schema)

sample_dataframe.write.format('delta').mode("overwrite").option('mergeSchema','true').saveAsTable("db_delta.sample_data")

將Cell類型切換為SQL，執行以下語句，查詢sample_data表中的資料。
```
SELECT * FROM db_delta.sample_data;
```

如果您想在AnalyticDB for MySQL控制台中使用Spark SQL分析sample_data表，可執行以下步驟。
1. 登入雲原生資料倉儲AnalyticDB MySQL控制台，在左上方選擇叢集所在地區。在左側導覽列，單擊集群清單。在集群清單上方，選擇產品系列，然後單擊目的地組群ID。
2. 在左側導覽列單擊作業開發 > SQL開發頁面，選擇Spark引擎和Interactive型資源群組。
3. 查詢sample_data資料。
```
SELECT * FROM db_delta.sample_data LIMIT 1000;
```

AnalyticDB：Notebook開發Spark作業

前提條件

使用限制

操作步驟

相關文檔