euler
euler copied to clipboard
alibaba
Euler2.0 分布式训练问题
你好！ 在跑分布式训练，参考了Euler 2.0 在大规模图上的应用 ，看到要把一个大的json数据分片成partition_num个json，分片的原则是边要和src 节点放在同一个json中，我没有找到这个分片的脚本，于是自己写了一个，分成多个json文件之后，需要转换成二进制文件，于是使用脚本
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)
设置了s=2在这个过程中发现会报错，查看原因出现在euler/euler/tools/json2partdat.py 的如下函数
self.nodes[edge_json['dst']].set_in_neighbor(
src_id,
edge_json['weight'],
edge_type,
self.gmeta.edge_type_count)
分析这个逻辑是设置self.nodes的这条边的dst的Node的邻居，然而self.nodes没有edge_json['dst']的Node，原因是我加载的这个grap.json_0 的文件只包含边和其src的节点，可能会出现其边的dst节点不在这个json_0文件中。
问题：
-
我在怀疑是否我对分片的含义存在理解错误，我是否有必要编写一个脚本将大的json文件按照“边和src节点分在同一个json文件中”这个规定来分片呢？
-
我看json2partdat.py 的代码逻辑好像就是分片的一个工作，但是输出的结果是.dat 文件，于是我困惑 shard_num 和 partition_num的含义。
-
假设已经完成上述的分片工作，接下来要启动euler
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:
我理解这里的shard_num 是和tf 的worker个数是一致的，euler的图引擎个数和tf worker个数一致，每个graph engine 加载一部分数据，我是否可以将分片好的数据存储在每个worker里而不是hdfs（目前只是想测试一下，数据使用的比较小），然后通过euler_graph_data_dir指定路径，另外 不是很明白zk_path的含义？
以上是我的问题，期待开发团队的回复
set_in_neighbor，在这版本里面没有用，你可以把这行注释掉 json2partdat.py 是将json明文数据转换成二进制数据，方便euler快速加载，shard_num 是比partition_num 更上层的概念，partition_num 是实际物理上的分片数，shard_num是euler加载的时候的逻辑分片数 zk_path 主要使用zk来同步euler上的meta信息
set_in_neighbor，在这版本里面没有用，你可以把这行注释掉 json2partdat.py 是将json明文数据转换成二进制数据，方便euler快速加载，shard_num 是比partition_num 更上层的概念，partition_num 是实际物理上的分片数，shard_num是euler加载的时候的逻辑分片数 zk_path 主要使用zk来同步euler上的meta信息
我按照如下的流程进行了工作： 1. 源码编译 euler2.0 2. 跑了run_gat.py 脚本，跑通了，该过程准备了数据，且将cora的原始数据转换成了一个convert_data.json文件 3. 我基于这个json文件，使用自己编写的脚本进行拆分成两个json文件（按照边和其src节点分配到同一个json文件的原则） 4. 接下来将各个json文件转换成二进制文件，按照你说的，我将set_in_neighbor注释掉。依次执行如下： 其中meta为一个只包含 {} 两个字符的文件， shard 设置为2，最后运行成功，在./temp文件中生成了三个文件Edge euler.meta Node, Edge 和Node文件下分别有0_0.dat 0_1.dat 1_0.dat 1_1.dat四个文件
sh gen_partitioned_data.sh cora/graph.json_0 cora/meta ./temp/ 2 0
sh gen_partitioned_data.sh cora/graph.json_1 cora/meta ./temp/ 2 1
- 我将上述temp文件中的内容拷贝到一个./euler 的文件中，该路径设置为data_dir。
- 然后我准备在本地开发机上模拟tf分布式，写了一个gat_dist.py脚本，其中tf culster的配置信息如下：
job_name = flag.job_name
index = flag.task_index
tf_config = {
'cluster': {'chief': [localhost:2223], 'worker': [ localhost:2224, localhost:2225], 'ps': [localhost:2222]},
'task': {'type': job_name, 'index': task_index}
}
然后按照启动euler：
data_dir = "./temp/euler"
shard_num = flag.task_index # 我的理解是tf 的worker数和shard_num一致，这样理解是否有误？？？还望解惑
zk_addr = "172.22.0.107:2181"
zk_path = "/tf_euler"
euler.start(
directory=data_dir, # 图数据路径 生成二进制的euler文件
shard_idx=task_index, # 当前启动的进程为k号shard
shard_num=shard_num, # 一共有N个shard
zk_addr=zk_addr, # Zookeeper address, ip:port
zk_path=zk_path, # Zookeeper path
module=euler.Module.DEFAULT_MODULE)
然后是init graph
if not work_name == 'ps':
tf_euler.initialize_graph({
'mode': 'remote',
'zk_server': zk_addr,
'zk_path': zk_path,
'shard_num': shard_num,
'num_retries': 1
})
我按照上面的跑了
euler/euler/core/graph/graph_builder.cc:169] Graph Edge Count:0
问题出在哪呢？
分布式生成数据这块应该是有个bug，我们会尽快解决
再问几个问题：
- 我跑的例子，是将切分好的数据直接放到tf的每个worker的机器上（directory = "./temp/euler", 通过类似TONY的分布式训练平台），而没有使用hdfs，我理解这种方式也是可以的吧。
- 在tf的模型分布式部署的时候，我的理解是： euler将完整的一张图分布存储在多个机器上，每个机器上启动了一个euler的graph engine，这个图引擎将该机器分到的子图全部加载到内存中。 一个上层的GNN算法，需要采样一个batch的数据，这个采样操作可能是多个采样op组成的一个图（比如是samplenode + sampleedge + sampleneighbor +...），该请求图被发送给C++后端的时候，会被解析拆分几个并行的op（指不存在前后依赖的op），然后去各个分布式机器上请求，最后汇总成一个batch数据，返回给模型做为输入。 那么是不是每个机器上的图引擎都将该机器分配到的子图都加载到了内存？还是加载了一部分？这个一部分是不是就是一个.dat ？ 此外这些图引擎所在的机器是不是同worker的机器是同一个，能不能比较详细一点的解答一下 shard_num 和tf_worker 和 分片 ，以及最后的.dat 文件，这几个概念，然后解释一下 euler2.0后端是一个什么结构吗？
分布式生成数据这块应该是有个bug，我们会尽快解决
这个bug 实在哪里，可以提前告知，我自己改一下，然后能不能出一个分布式部署某个算法的具体的文档例子啊
分布式生成数据这块应该是有个bug，我们会尽快解决
想借助euler2.0 来部署公司的分布式，对euler的技术 内幕不是很清楚，可以介绍一下吗，或者有没有博客介绍之类的
分布式生成数据这块应该是有个bug，我们会尽快解决
这个bug 实在哪里，可以提前告知，我自己改一下，然后能不能出一个分布式部署某个算法的具体的文档例子啊
目前分布式生成数据的问题是meta生成的不对，这个需要先全部过一遍数据生成meta，然后再过一遍数据生成dat文件 如果全量数据不大的话 可以参考这个https://github.com/alibaba/euler/wiki/Euler-2.0-%E6%95%B0%E6%8D%AE%E5%87%86%E5%A4%87#%E4%BD%BF%E7%94%A8build%E5%B7%A5%E5%85%B7%E7%94%9F%E6%88%90euler%E4%BA%8C%E8%BF%9B%E5%88%B6%E6%96%87%E4%BB%B6 把全量数据用一个json来处理
分布式生成数据这块应该是有个bug，我们会尽快解决
再问几个问题：
- 我跑的例子，是将切分好的数据直接放到tf的每个worker的机器上（directory = "./temp/euler", 通过类似TONY的分布式训练平台），而没有使用hdfs，我理解这种方式也是可以的吧。
- 在tf的模型分布式部署的时候，我的理解是： euler将完整的一张图分布存储在多个机器上，每个机器上启动了一个euler的graph engine，这个图引擎将该机器分到的子图全部加载到内存中。 一个上层的GNN算法，需要采样一个batch的数据，这个采样操作可能是多个采样op组成的一个图（比如是samplenode + sampleedge + sampleneighbor +...），该请求图被发送给C++后端的时候，会被解析拆分几个并行的op（指不存在前后依赖的op），然后去各个分布式机器上请求，最后汇总成一个batch数据，返回给模型做为输入。 对的
那么是不是每个机器上的图引擎都将该机器分配到的子图都加载到了内存？还是加载了一部分？这个一部分是不是就是一个.dat ？ 此外这些图引擎所在的机器是不是同worker的机器是同一个，能不能比较详细一点的解答一下 shard_num 和tf_worker 和 分片 ，以及最后的.dat 文件，这几个概念，然后解释一下 euler2.0后端是一个什么结构吗？
euler2.0的后端数据结构参考https://github.com/alibaba/euler/blob/c2a71faae59c1495b6dabcf6aec0acb4d93a7bb1/euler/core/graph/graph.h#L41
分布式生成数据这块应该是有个bug，我们会尽快解决
这个bug 实在哪里，可以提前告知，我自己改一下，然后能不能出一个分布式部署某个算法的具体的文档例子啊
目前分布式生成数据的问题是meta生成的不对，这个需要先全部过一遍数据生成meta，然后再过一遍数据生成dat文件 如果全量数据不大的话 可以参考这个https://github.com/alibaba/euler/wiki/Euler-2.0-%E6%95%B0%E6%8D%AE%E5%87%86%E5%A4%87#%E4%BD%BF%E7%94%A8build%E5%B7%A5%E5%85%B7%E7%94%9F%E6%88%90euler%E4%BA%8C%E8%BF%9B%E5%88%B6%E6%96%87%E4%BB%B6 把全量数据用一个json来处理
是的，我暂时想把你们的gat的例子跑成分布式，你的意思是，可以不用将gat用到的cora的json文件拆分成多个吗？我现在单机本地跑通了 gat 发现生成了一个convert_data.json 文件，意思我可以不用将 其分成多个json，而是直接将其转成二进制文件dat就可以了吗？然后我不把dat推到hdfs（暂时hdfs还有问题，想先跑通数据在本地的情况），而是在每个tf的worker机器上都保存一份数据，就可以使用euler.start()（是否ps chief 和worker 都要跑这个代码） 和tf_euler.initialize_grap()（是否只要worker和chief跑这个代码就好了）就将数据load到每个worker的内存中，并由euler的graph engine来管理其各自的子图了吗？
分布式生成数据这块应该是有个bug，我们会尽快解决
这个bug 实在哪里，可以提前告知，我自己改一下，然后能不能出一个分布式部署某个算法的具体的文档例子啊
目前分布式生成数据的问题是meta生成的不对，这个需要先全部过一遍数据生成meta，然后再过一遍数据生成dat文件 如果全量数据不大的话 可以参考这个https://github.com/alibaba/euler/wiki/Euler-2.0-%E6%95%B0%E6%8D%AE%E5%87%86%E5%A4%87#%E4%BD%BF%E7%94%A8build%E5%B7%A5%E5%85%B7%E7%94%9F%E6%88%90euler%E4%BA%8C%E8%BF%9B%E5%88%B6%E6%96%87%E4%BB%B6 把全量数据用一个json来处理
是的，我暂时想把你们的gat的例子跑成分布式，你的意思是，可以不用将gat用到的cora的json文件拆分成多个吗？ 是的 我现在单机本地跑通了 gat 发现生成了一个convert_data.json 文件，意思我可以不用将 其分成多个json，而是直接将其转成二进制文件dat就可以了吗？ 对 然后我不把dat推到hdfs（暂时hdfs还有问题，想先跑通数据在本地的情况），而是在每个tf的worker机器上都保存一份数据，就可以使用euler.start()（是否ps chief 和worker 都要跑这个代码） 和tf_euler.initialize_grap()（是否只要worker和chief跑这个代码就好了）就将数据load到每个worker的内存中，并由euler的graph engine来管理其各自的子图了吗？
euler 和 worker 是两个东西，euler是指一些机器专门负责load数据，提供查询服务。worker就是tf的worker的意思。所以数据是由euler来load的，load数据的那台机器才调用euler.start(). worker 调用 tf_euler.initialize_grap()，ps不用调用euler的任何接口 有几个dat文件，就可以起动几个euler，每个euler负责一份dat，如果只有一个dat，起一个euler就可以了
分布式生成数据我们会尽快给出完整的处理脚本
您好，请问现在分布式训练数据生成的bug修复了吗？
如果你的json文件只有一个的话，转换程序没有问题；
如果你的json文件有多个的话，那么生成的euler.meta的程序有bug, 具体的问题是统计的node_count和edge_count不准确，问题出在gen_partitioned_data.sh中传给json2meta.py的数据只有第一个part的json，而生成meta数据需要全图的数据。
这里提供一个简单的解决方法：
（1）修改gen_partitioned_data.sh如下图红框中的参数为存放所有json文件的目录
（2）修改json2meta.py的代码，见截图红框部分，经过（1）的修改，这个inputs当前的值为json文件所在的目录，接下来需要将inputs的值替换为该目录下的所有文件路径列表，这里通过实现一个函数即可：输入一个路径，返回包含该目录下所有文件路径的list
