Euler 2.0 在大规模图上的应用

在本章，我们将介绍euler2.0是如何在大规模图上应用。

在现实应用，尤其是工业级应用中，所面临的图是十分巨大的，比如在淘宝的应用场景中，往往有数亿个节点和数十亿条边。在这样一个规模的图下，单机的训练图神经网络是不可行的。如何处理大规模图数据是图神经网络能够真正落地的关键之一。

Euler-2.0支持图分割和高效稳定的分布式训练，可以轻松支撑如此规模的图数据。

本章将价绍如何使用Euler 2.0来处理图数据，进行分布式训练。

Note:

与其他场景应用的联系和区别的对比见这里。

数据准备

Euler2.0 大规模图数据生成

大规模图数据生成在步骤与一般图生成类似，也是分为两步：

1.构建图数据的JSON文件和Meta文件

一般而言，在大规模图下，由于图数据量很大，需要对对应的JSON文件进行切分，形成多个JSON文件，切分的原则是：边和其源节点（src节点）在同一个JSON文件中。假设JSON分片数目为partition_num=p，则需生成p个JSON文件，形如：

graph.json_0
graph.json_1
...
graph.json_(p-1)

对于meta文件而言，所有的JSON文件共用一个meta文件。

2.利用Euler 2.0的图数据转化工具将步骤1中的JSON文件和Meta文件转化成Euler2.0可以加载的二进制文件

为了生成相应的图数据二进制文件，Euler2.0提供了相应的脚本工具在项目根目录下的euler/tools/gen_partitioned_data.sh 可以将上述的多个JSON文件转化为Euler加载所需要的二进制文件。

假设JSON分片数目为partition_num=p，euler server分片数目为shard_num=s，那么依次执行

sh gen_partitioned_data.sh graph.json_0 index_meta output_dir s 0
sh gen_partitioned_data.sh graph.json_1 index_meta output_dir s 1
...
sh gen_partitioned_data.sh graph.json_(p-1) index_meta output_dir s (p-1)

其中：

graph.json_[index]：为各个切分过的euler json文件名
index_meta：全局公用的meta索引文件名，每次执行时包保持索引文件相同
output_dir：Euler二进制文件输出目录，每次执行后，转换工具会把转换的结果分别保存在同一个目录中，所以每次执行的时候要保持这个参数相同。
shard_num(s)：euler 二进制转换完成后的分片个数，每次执行的时候保持一致。
partition_num(p)：euler JSON文件的分片个数，每次执行的时候要根据实际的分片修改这个参数。

Euler2.0 大规模图数据加载

对于大规模分布式图数据加载而言一般需要以下三步：

在不同机器上，运行euler启动python脚本
初始化euler引擎
调用Euler op对节点或者边进行采样获邻居采样

在不同机器上，运行euler启动python脚本

假设数据被分割成了N个shards，那么在N台机器上运行下面的Euler启动python脚本。假设机器 0~~N-1 分别对应 0~~N-1个shard，那么第k个shard运行脚本为

euler.start(
    directory='euler_graph_data_dir', # 图数据路径
    shard_idx=k,     # 当前启动的进程为k号shard
    shard_num=N,     # 一共有N个shard
    zk_addr=zk_addr, # Zookeeper address, ip:port
    zk_path=zk_path, # Zookeeper path
    module=euler.Module.DEFAULT_MODULE)
while True:
    time.sleep(1)

初始化euler引擎

tf_euler.initialize_graph({
        'mode': 'remote',
        'zk_server': zk_addr,   # Zookeeper address, ip:port
        'zk_path': zk_path,     # Zookeeper path
        'shard_num': shard_num, # 一共有N个shard
        'num_retries': 1
    })

调用Euler op对节点或者边进行采样获邻居采样

启动并初始化euler引擎之后，后台图数据的分布式调度会自动启用，用户就可以和之前单机的方式一样去使用Euler相关的op，比如生成batch样本数据的时候，仍然可以使用sample_node()来实现。

tf_euler.sample_node(inputs, params['train_node_type'])

模型实现

在分布式的场景下，模型的实现部分仍然和单机的模式一样，不需要有任何的改动，具体实现可以参考进阶教程下的各个应用中的图模型的实现，在这里就不再展开介绍。

模型训练

加载图数据

在不同机器上运行euler启动的python脚本

比如：

start_euler.py

euler.start(
    directory='euler_graph_data_dir', # 图数据路径
    shard_idx=k,     # 当前启动的进程为k号shard
    shard_num=N,     # 一共有N个shard
    zk_addr=zk_addr, # Zookeeper address, ip:port
    zk_path=zk_path, # Zookeeper path
    module=euler.Module.DEFAULT_MODULE)
while True:
    time.sleep(1)

在模型训练的python中初始化euler引擎

#初始化euler引擎
if not task_type == 'ps':
    tf_euler.initialize_graph({
        'mode': 'remote',
        'zk_server': zk_addr,
        'zk_path': zk_path,
        'shard_num': shard_num,
        'num_retries': 1
    })

以上两步之后便可使用euler进行图数据操作

创建模型并训练

# 分布式参数配置
# 参考https://www.tensorflow.org/api_docs/python/tf/estimator/RunConfig进行配置
'''
cluster = {'chief': ['host0:2222'],
           'ps': ['host1:2222', 'host2:2222'],
           'worker': ['host3:2222', 'host4:2222']}
task_type = 'worker'
task_id = 0
os.environ['TF_CONFIG'] = json.dumps(
    {'cluster': cluster,
     'task': {'type': task_type, 'index': task_id}})
})
'''

#初始化euler引擎
if not task_type == 'ps':
    tf_euler.initialize_graph({
        'mode': 'remote',
        'zk_server': zk_addr,
        'zk_path': zk_path,
        'shard_num': shard_num,
        'num_retries': 1
    })

tf.logging.set_verbosity(tf.logging.INFO)

# 算法模型的构建，可以使用example中的算法模型，自行配置各个参数
model_cls = your_model(model_params)

# 训练参数配置，提供必要的训练参数
params = estimator_training_params

# Estimator的创建与训练/验证/预测
config = tf.estimator.RunConfig(log_step_count_steps=None)
base_estimator = NodeEstimator(model_cls, params, config)
base_estimator.train_and_evaluate()