标签: distributed

这是在我的机器上重现错误的代码：

import numpy as np
import xgboost as xgb
import dask.array as da
import dask.distributed
from dask_cuda import LocalCUDACluster
from dask.distributed import Client

X = da.from_array(np.random.randint(0,10,size=(10,10)))
Y = da.from_array(np.random.randint(0,10,size=(10,1)))

cluster = LocalCUDACluster(n_workers=4, threads_per_worker=1)
client = Client(cluster)

dtrain = xgb.dask.DaskDeviceQuantileDMatrix(client=client, data=X, label=Y)

params = {'tree_method':'gpu_hist','objective':'rank:pairwise','min_child_weight':1,'max_depth':3,'eta':0.1} 
watchlist = [(trainLong, 'train')] 
reg= xgb.dask.train(client, params, dtrain, num_boost_round=10,evals=watchlist,verbose_eval=1)

这是错误的摘要：

---------------------------------------------------------------------------
TypeError                                 Traceback (most recent call last)
<ipython-input-9-ff1b0329f2f9> in <module>
      1 params = {'tree_method':'gpu_hist','objective':'rank:pairwise','min_child_weight':1,'max_depth':3,'eta':0.1}
      2 watchlist = [(trainLong, 'train')]
----> 3 regLong = xgb.dask.train(client, params, trainLong, …

我知道关于在 Julia 中使用 @threads、@distributed 和其他方法运行并行 for 循环存在很多问题。我尝试在那里实施解决方案，但没有成功。我想做的结构如下。

for index in list_of_indices  
    data = h5read("data_set_$index.h5")  
    result = perform_function(data)  
    save(result)  
end

数据集是独立的，并且该循环的任何部分都不依赖于任何其他部分。看来这应该是可并行的。

我尝试过，例如

“@threads for index in list_of_indices...”并且出现分段错误

“@distributed for index in list_of_indices...”并且代码实际上并未对我的数据执行该功能。

我想我错过了一些关于并行进程如何工作的信息，任何见解将不胜感激。

这是一个 MWE：

假设我们的工作目录中有文件 data_1.h5、data_2.h5、data_3.h5。（我不知道如何使事情比这更独立，因为我认为问题是由要求多个线程读取文件引起的。）

using Distributed
using HDF5

list = [1,2,3]


Threads.@threads for index in list
    data = h5read("data_$index.h5", "data")
    println(data)
end

我得到的错误是

signal (11): Segmentation fault
signal (6): Aborted
Allocations: 1587194 (Pool: 1586780; Big: 414); GC: 1
Segmentation fault (core dumped)

如果有人拥有可以查询的点对点系统，则希望

• 减少整个网络上的查询总数（通过将“热门”项目和“类似”项目一起分布）
• 避免在每个节点上过多存储
• 确保在面对客户端停机，硬件故障以及用户离开时（即使是中等程度的稀有商品）也具有良好的可用性（可能为档案管理员/历史学家检测稀有商品）
• 避免查询在网络分区的情况下找不到匹配项

鉴于这些要求：

1. 有没有标准的方法？如果没有，是否有任何受人尊敬的实验研究？我对分配方案很熟悉，但是我还没有看到任何真正可以解决学习问题的方法。
2. 我是否缺少任何明显的标准？
3. 是否有人有兴趣解决/解决这个问题？（如果是这样，我很乐意将本周末聚集在一起的非常la脚的模拟器的一部分开源，并且通常会提供无用的建议）。

@cdv：我现在已经看过视频了，它非常棒，尽管我觉得它并没有完全适用于可插拔的发行策略，但肯定是90％的实现。但是，这些问题突出了这种方法的有用差异，这些差异解决了我的一些进一步担忧，并为我提供了后续的参考。因此，尽管我认为这个问题尚待解决，但我暂时接受您的回答。

我已经启动了很多任务,但其中一些还没有完成(763个任务),处于PENDING状态,但是系统没有处理任何东西......有可能重新执行这个任务给芹菜task_id吗？

ZooKeeper FAQ告诉您，当状态变为SESSION_EXPIRED时，必须重新创建ZooKeeper对象。

是否有代码示例可以正确进行这种重新连接？zookeeper发行版中的Recipes文件夹包含两个代码示例，两个示例均假定会话永不过期。这是幼稚的，因为在服务器与ZooKeeper群集断开连接5分钟之后，重新建立连接后，队列和锁定都将不再起作用。

我正在重构一个将进行大量计算的分析系统,我需要一些关于可能的架构设计的想法,以解决我面临的数据一致性问题.

当前架构

我有一个基于队列的系统,其中不同的请求应用程序创建最终由工作者使用的消息.

每个" 请求应用程序 "将大型计算细分为较小的部分,这些部分将被发送到队列并由工作人员处理.

当所有部分都完成后,原始的"请求应用程序"将合并结果.

此外,工作者使用来自集中式数据库(SQL Server)的信息来处理请求(重要:工作人员不会更改数据库上的任何数据,只会消耗它).

问题

好.到现在为止还挺好.当我们包含更新数据库信息的Web服务时,就会出现问题.这可能在任何时候发生,但至关重要的是,源自相同"请求应用程序"的每个"大计算"在数据库上看到相同的数据.

例如:

1. 应用A生成消息A1和A2,将其发送到队列
2. 工人W1选择消息A1进行处理.
3. Web服务器更新数据库,从状态S0更改为S1.
4. 工人W2拿起消息A2进行处理

我不能让工人W2使用数据库的状态S1.为了使整个计算保持一致,它应该使用先前的S0状态.

思考

1. 一种锁定模式,用于防止Web服务器在工作者消耗数据时更改数据库.

   • 缺点:锁定可能会长时间打开,因为不同的"请求应用程序"的计算形式可能会重叠(A1,B1,A2,B2,C1,B3等).

2. 在数据库和worker(通过req.app控制db缓存的服务器)之间创建新层

   • 缺点:添加另一个层可能会带来很大的开销(可能？),这是很多工作,因为我将不得不重写工作者的持久性(很多代码).

我正在等待第二个解决方案,但对此并不十分自信.

有什么好主意吗？我设计错了,还是错过了什么？

OBS:

• 这是一个巨大的2层遗留系统(在C#中),我们正在尝试尽可能少地努力发展成更具可扩展性的解决方案.
• 每个worker都可能在不同的服务器上运行.

我在apache spark中遇到了一个奇怪的问题,我很感激一些帮助.从HDFS读取数据(并从JSON做一些转换为一个对象)在下一阶段(处理所述对象)之后2个分区已经(在总总分512)被处理失败之后.这种情况发生在大型数据集上(我注意到的最小数据集大约为700兆,但可能更低,我还没有缩小它).

编辑:700兆是tgz文件大小,未压缩它是6演出.
编辑2:火花1.1.0也发生了同样的事情

我在32核,60 gig机器上使用本地主机运行spark,具有以下设置:

spark.akka.timeout = 200
spark.shuffle.consolidateFiles = true
spark.kryoserializer.buffer.mb = 128
spark.reducer.maxMbInFlight = 128

16 gig执行器堆大小.内存没有被最大化,CPU负载可以忽略不计.Spark永远挂起.

以下是火花日志:

14/09/11 10:19:52 INFO HadoopRDD: Input split: hdfs://localhost:9000/spew/data/json.lines:6351070299+12428842
14/09/11 10:19:53 INFO Executor: Serialized size of result for 511 is 1263
14/09/11 10:19:53 INFO Executor: Sending result for 511 directly to driver
14/09/11 10:19:53 INFO Executor: Finished task ID 511
14/09/11 10:19:53 INFO TaskSetManager: Finished TID 511 in 868 ms on localhost (progress: 512/512)
14/09/11 10:19:53 INFO DAGScheduler: Completed ShuffleMapTask(3, 511) …

我正在测试majordomo经纪人的吞吐量.随github上的majordomo代码附带的test_client.c发送同步请求.我想测试majordomo代理可以实现的最大吞吐量.规范(http://rfc.zeromq.org/spec:7)表示它每秒可以切换多达一百万条消息.

首先,我更改了客户端代码以异步发送100k请求.即使在所有套接字上的HWM设置得足够高并且将TCP缓冲区增加到4 MB之后,我也会观察到三个并行运行的客户端丢包.

所以我改变了客户端一次发送10k个请求,然后为它收到的每个回复发送两个请求.我选择10k是因为这允许我并行运行多达十个客户端(每个发送100k消息)而不会丢失任何数据包.这是客户端代码:

#include "../include/mdp.h"
#include <time.h>
int main (int argc, char *argv [])
{
    int verbose = (argc > 1 && streq (argv [1], "-v"));
    mdp_client_t *session = mdp_client_new (argv[1], verbose);
    int count1, count2;
    struct timeval start,end;
    gettimeofday(&start, NULL);
    for (count1 = 0; count1 < 10000; count1++) {
        zmsg_t *request = zmsg_new ();
        zmsg_pushstr (request, "Hello world");
        mdp_client_send (session, "echo", &request);
    }
    for (count1 = 0; count1 < 45000; count1++) {
        zmsg_t *reply = mdp_client_recv (session,NULL,NULL); …

我尝试以分布式方式运行一个简单的应用程序来测试故障转移接管功能但是失败了.

我想要的是:

应用程序myapp_api使用rest api,它将myapp应用程序作为依赖项.我想从myapp_api3个节点开始,我希望整个应用程序(myapp_api+ myapp)只能同时在一个节点上工作.

怎么了:

主app(myapp_api)按预期工作:仅在具有故障转移和接管的一个节点上.但由于某些原因,依赖myapp总是从每个节点开始.我希望它只能同时在一个节点上工作.

我所做的:

我以第一个节点的配置为例.

[
    {kernel,
    [{distributed, [{myapp_api,
        1000,
        ['n1@myhost', {'n2@myhost', 'n3@myhost'}]}]},
        {sync_nodes_optional, ['n2@myhost', 'n3@myhost']},
        {sync_nodes_timeout, 5000}
    ]}
].

我erl -sname nI -config nI.config -pa apps/*/ebin deps/*/ebin -s myapp_api在每个节点打电话 .

我计划基于微服务构建后端。下图展示了我当前的想法：

两个重要功能是：

• 上载大型文字和/或视频文件
• 流视频-在Web应用程序和本机移动应用程序中显示用户

最终没有确定技术栈，但最初我想到的是：

• 网路应用程式-ReactJs / Angular
• 后端应用程序-Ruby on Rails

我对当前的概念有以下疑问：

1. API Gateway是否应该充当路由器，将用户的请求重定向到特定的微服务？或者...应该是带有API的专用应用程序（例如Ruby on Rails应用程序）吗？
2. 如何进行授权？我应该为此使用单独的微服务吗？假设用户上传文件，而他的请求应该转到第三个微服务：“大数据上传”。我应该在何时何地授权他访问？在那个微服务或更早的API Gateway中？还是应该在“身份验证微服务”中进行身份验证？
3. 上传大文件-假设用户想通过HTTP将大型文件（带有原始数据的视频或压缩文本文件）从移动应用程序传输到后端。他的请求转到API网关，然后将其重定向到“大数据上传”微服务。应用程序将文件保存到对象存储。是上传文件的正确路径吗？或者我可以做一些捷径来缩短文件的路径？
4. 视频流-用户上传视频文件时，我想将其放入资产（对象存储-例如Amazon S3）。将此视频呈现给Web应用程序或移动应用程序中的用户是否足够？（除了代码转换器服务和CDN之外）
5. 负载平衡-使用负载平衡来控制流向微服务实例的流量（在API网关和绿色微服务之间的图片上）是否合理？还是这不是一个好方法，因为我们可能会丢失一些有关请求/收件人/用户的信息，甚至API网关都会成为更严重的“瓶颈”？
6. 您认为这种架构概念是否具有易于扩展的良好潜力？当然，省略了硬件和软件配置。