在 AKS 上的 HDInsight 中的 Apache Spark™ 群集上提交和管理作业

项目
09/20/2024

重要

AKS 上的 Azure HDInsight 已于 2025 年 1 月 31 日停用。了解此公告的详细信息。

需要将工作负荷迁移到 Microsoft Fabric 或等效的 Azure 产品，以避免工作负荷突然终止。

重要

此功能目前以预览版提供。 Microsoft Azure 预览版补充使用条款包括适用于测试版、预览版或尚未正式发布的 Azure 功能的更多法律条款。有关此特定预览的信息，请参阅有关在 AKS 上的 Azure HDInsight 的预览信息。有关问题或功能建议，请在 AskHDInsight 上提交请求，并提供详细信息，并关注我们以获取更多关于 Azure HDInsight 社区的更新。

创建群集后，用户可以使用各种界面来提交和管理作业

使用 Jupyter
使用 Zeppelin 软件
使用 ssh 运行 spark-submit

使用 Jupyter

先决条件

基于 AKS 的 HDInsight 平台上的 Apache Spark™ 群集。有关详细信息，请参阅创建 Apache Spark 群集。

Jupyter Notebook 是支持各种编程语言的交互式笔记本环境。

创建 Jupyter Notebook

导航到 Apache Spark™ 群集页并打开“概述” 选项卡。单击 Jupyter，它会要求你进行身份验证并打开 Jupyter 网页。
在 Jupyter 网页中，选择“新建”> PySpark 以创建笔记本。

新创建并打开的笔记本，名称为 Untitled(Untitled.ipynb)。

注意

通过使用 PySpark 或 Python 3 内核创建笔记本，在运行第一个代码单元时，系统会自动创建 Spark 会话。无需显式创建会话。

将以下代码粘贴到 Jupyter Notebook 的空单元格中，然后按 Shift + ENTER 运行代码。有关 Jupyter 上的更多控件，请参阅此处。

%matplotlib inline
import pandas as pd
import matplotlib.pyplot as plt
data1 = [22,40,10,50,70]
s1 = pd.Series(data1)   #One-dimensional ndarray with axis labels (including time series).

data2 = data1
index = ['John','sam','anna','smith','ben']
s2 = pd.Series(data2,index=index)

data3 = {'John':22, 'sam':40, 'anna':10,'smith':50,'ben':70}
s3 = pd.Series(data3)

s3['jp'] = 32     #insert a new row
s3['John'] = 88

names = ['John','sam','anna','smith','ben']
ages = [10,40,50,48,70]
name_series = pd.Series(names)
age_series = pd.Series(ages)

data_dict = {'name':name_series, 'age':age_series}
dframe = pd.DataFrame(data_dict)   
#create a pandas DataFrame from dictionary

dframe['age_plus_five'] = dframe['age'] + 5   
#create a new column
dframe.pop('age_plus_five')
#dframe.pop('age')

salary = [1000,6000,4000,8000,10000]
salary_series = pd.Series(salary)
new_data_dict = {'name':name_series, 'age':age_series,'salary':salary_series}
new_dframe = pd.DataFrame(new_data_dict)
new_dframe['average_salary'] = new_dframe['age']*90

new_dframe.index = new_dframe['name']
print(new_dframe.loc['sam'])

绘制一张图，将“工资”作为X轴，“年龄”作为Y轴。
在同一笔记本中，将以下代码粘贴到 Jupyter Notebook 的空单元格中，然后按 SHIFT + ENTER 运行代码。
```
%matplotlib inline
import pandas as pd
import matplotlib.pyplot as plt

plt.plot(age_series,salary_series)
plt.show()
```

保存笔记本

在笔记本菜单栏中，导航到“文件”>“保存”和“检查点”。
关闭笔记本以释放群集资源：在笔记本菜单栏中，导航到“文件 > 关闭并停止。您还可以在示例文件夹下运行任何笔记本文件。

使用 Apache Zeppelin 笔记本

在 AKS 上的 HDInsight 中，Apache Spark 群集包括 Apache Zeppelin 笔记本。使用笔记本运行 Apache Spark 作业。本文介绍如何在 AKS 群集上的 HDInsight 上使用 Zeppelin 笔记本。

先决条件

AKS 上的 HDInsight 上的 Apache Spark 群集。有关说明，请参阅创建 Apache Spark 群集。

启动 Apache Zeppelin 笔记本

导航到 Apache Spark 群集概述页，然后从群集仪表板中选择 Zeppelin 笔记本。它会提示进行身份验证并打开 Zeppelin 页面。
创建新的笔记本。在标题窗格中，导航到笔记本 > 创建新笔记。确保笔记本标题栏显示为已连接状态。它表示右上角的绿色点。

在 Zeppelin Notebook 中运行以下代码：

%livy.pyspark
import pandas as pd
import matplotlib.pyplot as plt
data1 = [22,40,10,50,70]
s1 = pd.Series(data1)   #One-dimensional ndarray with axis labels (including time series).

data2 = data1
index = ['John','sam','anna','smith','ben']
s2 = pd.Series(data2,index=index)

data3 = {'John':22, 'sam':40, 'anna':10,'smith':50,'ben':70}
s3 = pd.Series(data3)

s3['jp'] = 32     #insert a new row

s3['John'] = 88

names = ['John','sam','anna','smith','ben']
ages = [10,40,50,48,70]
name_series = pd.Series(names)
age_series = pd.Series(ages)

data_dict = {'name':name_series, 'age':age_series}
dframe = pd.DataFrame(data_dict)   #create a pandas DataFrame from dictionary

dframe['age_plus_five'] = dframe['age'] + 5   #create a new column
dframe.pop('age_plus_five')
#dframe.pop('age')

salary = [1000,6000,4000,8000,10000]
salary_series = pd.Series(salary)
new_data_dict = {'name':name_series, 'age':age_series,'salary':salary_series}
new_dframe = pd.DataFrame(new_data_dict)
new_dframe['average_salary'] = new_dframe['age']*90

new_dframe.index = new_dframe['name']
print(new_dframe.loc['sam'])

点击段落中的播放按钮以运行代码片段。段落右侧的状态应从 READY、PENDING、RUNNING 到 FINISHED 进行。输出显示在同一段落底部。屏幕截图如下图所示：

输出：

使用 Spark 提交作业

使用以下命令“#vim samplefile.py”创建文件
此命令将打开 vim 文件

将以下代码粘贴到 vim 文件中

import pandas as pd
import matplotlib.pyplot as plt

From pyspark.sql import SparkSession
Spark = SparkSession.builder.master('yarn').appName('SparkSampleCode').getOrCreate()
# Initialize spark context
data1 = [22,40,10,50,70]
s1 = pd.Series(data1)   #One-dimensional ndarray with axis labels (including time series).

data2 = data1
index = ['John','sam','anna','smith','ben']
s2 = pd.Series(data2,index=index)

data3 = {'John':22, 'sam':40, 'anna':10,'smith':50,'ben':70}
 s3 = pd.Series(data3)

s3['jp'] = 32     #insert a new row

s3['John'] = 88

names = ['John','sam','anna','smith','ben']
ages = [10,40,50,48,70]
name_series = pd.Series(names)
age_series = pd.Series(ages)

data_dict = {'name':name_series, 'age':age_series}
dframe = pd.DataFrame(data_dict)   #create a pandas DataFrame from dictionary

dframe['age_plus_five'] = dframe['age'] + 5   #create a new column
dframe.pop('age_plus_five')
#dframe.pop('age')

salary = [1000,6000,4000,8000,10000]
salary_series = pd.Series(salary)
new_data_dict = {'name':name_series, 'age':age_series,'salary':salary_series}
new_dframe = pd.DataFrame(new_data_dict)
new_dframe['average_salary'] = new_dframe['age']*90

new_dframe.index = new_dframe['name']
print(new_dframe.loc['sam'])