在远程 Apache Hadoop 集群上使用Jupyter运行机器学习模型 Python

您可轻松利用远程 Hadoop 集群上的分布式Spark计算来构建和训练模型，同时确保对数据的安全访问。集群 Hadoop 的访问权限应通过服务 Execution Engine for Hadoop 配置至该 Hadoop 集群。您可以使用 Jupyter Enterprise Gateway (JEG) 或 Livy 访问 Hadoop Spark 来构建和训练模型。

您可以执行的任务：

使用 Python 环境
Hadoop 实用程序库
使用 JEG 构建模型
使用 Livy 构建模型
高可用性
负载均衡

使用 Python 环境

对于JEG和 Livy ，在构建模型时 Apache Hadoop ，您可使用 Watson Studio 中提供的模型框架。管理员本应将 Python 环境的镜像推送到 Hadoop 集群。

Hadoop 实用程序库

hadoop_utils_lib 和 hi_core_utils 库提供了许多有用的函数，这些函数可在 Jupyter 笔记本中调用，有助于进行模型操作。有关详细信息，请参阅《使用 Hadoop 实用程序（适用于 Python ）》。

使用 JEG 构建模型

在使用 JEG 构建模型之前，您必须创建一个 Hadoop 环境。环境设置控制：

目标系统： Hadoop
已推送用于执行的 Python 环境
用于提交 Spark YARN 作业的 YARN 队列
分配给该会话的 Spark 资源由环境中的设置控制

请执行以下任务，使用 JEG 构建和训练模型：

启动一个 Jupyter 笔记本

打开或创建一个 Jupyter 笔记本。在“选择运行时”下拉菜单中，选择“ Hadoop ”环境。这将使用您的用户身份安全地启动 JEG 内核，因此对 Hadoop 集群上数据和资源的访问将根据您的授权进行限制。

若您已升级 Cloud Pak for Data 至版本 4.7.0、 4.7.1 或 4.7.2 ，则必须在内核启动后，将以下命令作为首个单元格执行：

from pyspark.sql import SparkSession
from pyspark import SparkContext
spark = SparkSession.builder.getOrCreate()
sc = SparkContext.getOrCreate()

使用 shell 操作，您可以在 YARN 节点管理器上运行命令。例如，你可以使用以下命令：

!hostname -f
!hdfs dfs -ls /user

数据处理

您可以通过 Spark 会话从文件或 Hive HDFS 表中读取数据。

HDFS 文件

	df_data_1 = spark.read.format(
	    "org.apache.spark.sql.execution.datasources.csv.CSVFileFormat").option(
	    "header", "true").option("inferSchema", "true").load("hdfs:///user/cars.csv)

Hive 表

	df_data_1 = spark.sql("select * from cars")

您可以对 Spark DataFrame 执行操作，以转换和处理数据。

构建并训练模型

您可以将数据划分为训练集和验证集，并选择模型框架和算法来构建和评估模型。

将模型保存到 HDFS

您可以通过调用hi_core_utils库中的 HDFS Python函数将模型保存至。 hi_core_utils.write_model_to_hdfs(model=model, model_name="modelname")

此调用输出模型在. 上的保存位置 HDFS。该模型的 tar.gz 存档文件也将保存在该位置。

将模型导入 Watson Studio

要传输保存在 HDFS 上的模型，您应打开一个使用默认环境的笔记本，并调用 Hadoop 实用函数，将文件从 HDFS 传输到 Watson Studio。

hdfs_model_dir = "/user/{your username}/.dsxhi/models/{modelname}/{modelversion}/model.tar.gz"
wsl_model_path = "/project_data/data_asset/model.tar.gz"
hadoop_lib_utils.download_hdfs_file(webhdfs_endpoint, hdfs_model_dir, wsl_model_path)

使用 Livy 构建模型

启动一个 Jupyter 笔记本

打开或创建一个 Jupyter 笔记本。在“选择运行时”下拉菜单中，选择“默认 3.10Python ”环境。请注意，您不应选择 Hadoop 环境。

建立 Livy 遠端連線

您应导入 Python hadoop_lib_utils 库并调用实用 get_dsxhi_info 方法，以获取 Watson Studio 集群中已 Hadoop 注册系统的信息、可用服务以及已推送至 Python 环境的信息 HDFS。请使用以下命令：

HI_SYSTEMS = hadoop_lib_utils.get_dsxhi_info(showSummary=True)

您可以创建一个配置对象，用于指定其他 Spark 配置，例如 YARN 队列、驱动程序内存、执行器内存以及执行器数量。请使用以下命令：

myConfig={
 "queue": "default",
 "driverMemory": "1G",
 "numExecutors": 2
};

您应调用该 setup_livy_sparkmagic 函数来设置认证、 Hadoop 集群信息以及连接 Livy 的附加配置。如果您想使用已推送的 Python 环境，可以在参数 imageId 中指定：

HI_CONFIG = hadoop_lib_utils.setup_livy_sparkmagic(
  system="systemName	",
  livy="livyspark2",
  addlConfig=myConfig,
  imageId="jupyter-231n-py")

# (Re-)load sparkmagic to apply the new configs.
%reload_ext sparkmagic.magics
To create a Livy session, run
session_name = mysession
livy_endpoint = HI_CONFIG['LIVY']
%spark add -s $session_name -l python -k -u $livy_endpoint

这将使用您的用户身份安全地启动一个 YARN 应用程序，因此对 Hadoop 集群上数据和资源的访问将根据您的授权进行限制。如果发现调用返回了 500 错误，请检查 Hadoop 集群上的 Resource Manager 界面。如果 YARN 应用程序已成功创建，您可以忽略该 500 错误。

建立会话 Livy 后，要在集群 Hadoop 上运行代码，单元格的第一行应包含 %%spark -s $session_name。不以开头的单元在 %%spark 集群 Watson Studio 上本地运行。

数据处理

您可以通过 Spark 会话从文件或 Hive HDFS 表中读取数据。

HDFS 文件

%%spark -s $session_name
df_data_1 = spark.read.format(
    "org.apache.spark.sql.execution.datasources.csv.CSVFileFormat").option(
    "header", "true").option("inferSchema", "true").load("hdfs:///user/cars.csv)

Hive 表

%%spark -s $session_name
df_data_1 = spark.sql("select * from cars")

您可以对 Spark DataFrame 执行操作，以转换和处理数据。

构建并训练模型

您可以将数据划分为训练集和验证集，并选择模型框架和算法来构建和评估模型。

将模型保存到 HDFS

您可以通过调用hi_core_utils库中的 HDFS Python函数将模型保存至，具体调用方式如下：

%%spark -s $session_name
hi_core_utils.write_model_to_hdfs(model=model, model_name="modelname")

此调用输出模型在. 上的保存位置 HDFS。

将模型导入 Watson Studio

您可以使用 Hadoop 实用函数，将模型 HDFS 转换为 Watson Studio。请注意，该单元应在集群 Watson Studio 本地运行，且不应以 %%spark.

hdfs_model_dir = "/user/{your username}/.dsxhi/models/{modelname}/{modelversion}/model.tar.gz"
wsl_model_path = "/project_data/data_asset/model.tar.gz"
hadoop_lib_utils.download_hdfs_file(webhdfs_endpoint, hdfs_model_dir, wsl_model_path)

清理 Livy 会话

清理 Spark 会话 LivyHadoop 以释放远程 Hadoop 系统上占用的 YARN 资源，通过调用 %spark cleanup.

高可用性

任何位于故障节点上的活跃 JEG 或 Livy 会话都必须重新启动并再次运行。

负载均衡

JEG 和 Livy 的会话采用粘性会话分配机制，并遵循主动/被动模式。一个会话将在同一台 Execution Engine for Apache Hadoop 服务节点上运行，直到检测到故障；此时，所有新会话都将分配给下一台可用的 Execution Engine for Apache Hadoop 服务节点。