Ray是由加州大学伯克利分校 RISELab开源的新兴人工智能应用的分布式框架。它实现了一个统一的接口、分布式调度器、分布式容错存储，以满足高级人工智能技术对系统最新的、苛刻的要求。Ray 允许用户轻松高效地运行许多新兴的人工智能应用，例如，使用 RLlib 的深度强化学习、使用 Ray Tune 的可扩展超参数搜索、使用 AutoPandas 的自动程序合成等等。
在本文中，我们将介绍 RayOnSpark，这是新近添加到 Analytic Zoo 的功能之一。Analytics Zoo是开源的端到端数据分析 + 人工智能平台。RayOnSpark 允许用户直接在 Apache Hadoop/YANE 上运行 Ray 程序，这样用户就可以在现有的大数据集群上以分布式的方式轻松尝试各种新兴的人工智能应用。此外，大数据应用和人工智能应用并没有运行在两个独立的系统上，因为这往往会带来昂贵的数据传输成本和较高的端到端学习延迟。RayOnSpark 允许 Ray 应用无缝集成到 Apache Spark 数据处理管道中，并直接在内存中的 Spark RDD 或 DataFrame 上运行。
接下来，我们将重点阐述如何在 Hadoop/YARN 之上使用 PySpark 运行 Ray 集群和程序（见下面的图 1）。注意，虽然本文只展示了如何在 YARN 集群上运行 Ray，但同样的逻辑也可以应用于 Kubernetes 和 Apache Mesos。
图1 在Aparche Spark上部署Ray 为了说明预期的 RayOnSpark 工作流，我们将使用一个简单的 Ray 示例，它使用 Actor 收集服务器的 IP 并在 YARN 集群上运行该实例。
1、请按照下面的链接来安装 Anaconda： https://docs.conda.io/projects/conda/en/latest/user-guide/install/index.html
2、创建名为“zoo”（或任何其他名称）的虚拟环境，如下所示：
conda create -n zoo python=3.6
source activate zoo
3、将 Spark、Analytics Zoo、Jupyter 和 Ray 安装到 Conda 环境中。
source activate zoo
pip install analytics-zoo==0.6.0.dev6 (or above version)
pip install pyspark==2.4.3
pip install ray
conda install jupyter
4、安装 Java 环境。
condainstall -c anaconda openjdk=8.0.152 Spark
需要Java 环境设置。如果环境变量JAVA_HOME 已经用 JDK8 设置好，则可以跳过这一步。
5、搜索并记住 Hadoop 配置文件夹的路径，这是稍后在 YARN 上初始化 Spark 所需的。文件夹的层次结构如下所示：
6、启动 Jupyter Notebook。
jupyter notebook
7、在 Jupyter Notebook 中，只需调用 Analytics Zoo 提供的“init_spark_on_yarn” Python 方法，就可以在 YARN 上启动 SparkContext：
from zoo import init_spark_on_yarn
sc = init_spark_on_yarn(
hadoop_conf=path to hadoop_conf,
conda_name="zoo", # The name of your conda environment
num_executor=3,
executor_cores=4,
executor_memory="2g",
driver_memory="2g",
driver_cores=4,
extra_executor_memory_for_ray="3g")
注：在YARN上使用PySpark时，用户面临的一个挑战是，在集群中的每个节点上准备Python环境，而不修改集群。你可能会考虑使用rsync手动将依赖项从驱动程序转移到集群，但这需要时间，且容易出错。此外，你可能在生产环境中没有ssh权限。在这里，我们通过利用conda-pack和YARN分布式缓存来解决这个问题，以便帮助用户在集群中自动捆绑和分发Python依赖项。
8、在 YARN 上使用 PySpark 启动 Ray 集群。
import ray
from zoo.ray.util.raycontext import RayContex
ray_ctx = RayContext(sc=sc, object_store_memory="2g")
ray_ctx.init()
在RayOnSpark中，我们首先创建一个SparkContext，它将负责通过“ray start”在底层集群（即YARN容器）中启动Ray进程。对于每个Spark执行器，都会创建一个“Ray Manager”（见下面的图2）来管理Ray进程；当故障发生或者程序退出时，它将自动关闭或重新启动进程。
图2 在Aparche Spark上启动Ray的过程
“RayContext”是触发Ray集群部署的入口点。下面是调用“ray_ctx.init()”幕后的逻辑：
- 将在本地节点上启动一个Ray“驱动程序”。
- 带有Redis进程的单个Ray“master”将在一个Spark执行器上启动。
- 对于每个剩余的Spark执行器，将启动一个“Slave”Reylet。
- Raymaster和Raylet进程将配置为使用由“executor_cores”参数指定的内核数。
9、之后，我们将编写一些简单的代码来测试 Ray 集群是否已经成功启动。例如，以下代码将创建 Actors 来从分配的 YARN 容器中收集 IP。
@ray.remote
classTestRay():
def hostname(self):
import socket
return socket.gethostname()def ip(self):
import ray.services as rservices
return rservices.get_node_ip_address()actors =[TestRay.remote() for i in range(0, slave_num)]
print([ray.get(actor.hostname.remote())for actor in actors])
print([ray.get(actor.ip.remote())for actor in actors])
ray_ctx.stop()
10、在阅读上面的简单示例代码之后，你可以参考更复杂的 RayOnSpark 的 Jupyter Notebook ，这是基于实现 Sharded 参数服务器的官方 Ray 执行。
借助 Analytics Zoo 中的 RayOnSpark 支持，用户只需在 Ray 程序的顶部添加三行额外的 Python 代码（如下所示）：
sc = init_spark_on_yarn( … )
ray_ctx = RayContext(sc=sc, … )
ray_ctx.init( … )
...
这样就可以在现有的 Hadoop/YARN 集群中直接运行构建在 Ray 之上的人工智能新的应用，这些应用可以无缝集成到 Spark 数据处理管道中。作为第一个用力，我们目前正在使用 RayOnSpark 来实现自动机器学习对时间序列预测的支持（包括自动特征生成、模型选择和超参数调优）。
