distributed-computing - 如何计算分布式数据的均值？

Question

我如何在分布式计算中计算大型向量(系列)的算术平均值，我在多个节点上对数据进行分区。我不想使用 map reduce 范例。除了对每个节点上的单个总和进行简单计算然后将结果带到主节点并除以向量(系列)的大小之外，是否有任何分布式算法可以有效地计算平均值。

Answer 1

分布式平均共识是一种替代方法。

master 的 map-reduce 简单方法的问题是，如果你有大量的数据，本质上是为了让一切都相互依赖，计算数据可能需要很长时间，通过哪个时间的信息非常过时，因此是错误的，除非您锁定整个数据集——这对于大量分布式数据集来说是不切实际的。使用分布式平均共识(相同的方法适用于均值的替代算法)，您可以在不锁定数据的情况下实时获得更新、更好的均值当前值猜测。这是一篇关于它的论文的链接，但它的数学很重: http://web.stanford.edu/~boyd/papers/pdf/lms_consensus.pdf你可以用谷歌搜索很多关于它的论文。

一般概念是这样的:假设在每个节点上都有一个套接字监听器。您评估本地总和和平均值，然后将其发布到其他节点。每个节点监听其他节点，并在合理的时间范围内接收它们的总和和平均值。然后，您可以通过 (sumForAllNodes(storedAverage[node] * storedCount[node])/(sumForAllNodes(storedCount[node])) 评估总平均值的良好猜测。如果您有一个真正大的数据集，您可以只听新的存储在节点中的值，并修改本地计数和平均值，然后发布它们。

如果这花费的时间太长，您可以对每个节点中的随机数据子集进行平均。

这里有一些 c# 代码可以让您了解(使用 fleck 可以在比仅适用于 Windows 10 的 Microsoft WebSockets 实现更多版本的 Windows 上运行)。在两个节点上运行它，一个带有

<appSettings>
    <add key="thisNodeName" value="UK" />
</appSettings>

在 app.config 中，在另一个中使用“EU-North”。这是一些示例代码。这两个实例交换意味着使用 websockets。您只需要添加数据库的后端枚举。

using Fleck;

namespace WebSocketServer
{
    class Program
    {
        static List<IWebSocketConnection> _allSockets;
        static Dictionary<string,decimal> _allMeans;
        static Dictionary<string,decimal> _allCounts;
        private static decimal _localMean;
        private static decimal _localCount;
        private static decimal _localAggregate_count;
        private static decimal _localAggregate_average;

        static void Main(string[] args)
        {
            _allSockets = new List<IWebSocketConnection>();
            _allMeans = new Dictionary<string, decimal>();
            _allCounts = new Dictionary<string, decimal>();

            var serverAddresses = new Dictionary<string,string>();
            //serverAddresses.Add("USA-WestCoast", "ws://127.0.0.1:58951");
            //serverAddresses.Add("USA-EastCoast", "ws://127.0.0.1:58952");
            serverAddresses.Add("UK", "ws://127.0.0.1:58953");
            serverAddresses.Add("EU-North", "ws://127.0.0.1:58954");
            //serverAddresses.Add("EU-South", "ws://127.0.0.1:58955");
            foreach (var serverAddress in serverAddresses)
            {
                _allMeans.Add(serverAddress.Key, 0m);
                _allCounts.Add(serverAddress.Key, 0m);
            }

            var thisNodeName = ConfigurationSettings.AppSettings["thisNodeName"];   //for example "UK"
            var serverSocketAddress = serverAddresses.First(x=>x.Key==thisNodeName);
            serverAddresses.Remove(thisNodeName);

            var websocketServer = new Fleck.WebSocketServer(serverSocketAddress.Value);

            websocketServer.Start(socket =>
            {
                socket.OnOpen = () =>
                {
                    Console.WriteLine("Open!");
                    _allSockets.Add(socket);
                };
                socket.OnClose = () =>
                {
                    Console.WriteLine("Close!");
                    _allSockets.Remove(socket);
                };
                socket.OnMessage = message =>
                {
                    Console.WriteLine(message + " received");

                    var parameters = message.Split('~');
                    var remoteHost = parameters[0];
                    var remoteMean = decimal.Parse(parameters[1]);
                    var remoteCount = decimal.Parse(parameters[2]);
                    _allMeans[remoteHost] = remoteMean;
                    _allCounts[remoteHost] = remoteCount;


                };
            });
            while (true)
            {
                //evaluate my local average and count
                Random rand = new Random(DateTime.Now.Millisecond);
                _localMean = 234.00m + (rand.Next(0, 100) - 50)/10.0m;
                _localCount = 222m + rand.Next(0, 100);

                //evaluate my local aggregate average using means and counts sent from all other nodes
                //could publish aggregate averages to other nodes, if you wanted to monitor disagreement between nodes
                var total_mean_times_count = 0m;
                var total_count = 0m;
                foreach (var server in serverAddresses)
                {
                    total_mean_times_count += _allCounts[server.Key]*_allMeans[server.Key];
                    total_count += _allCounts[server.Key];
                }
                //add on local mean and count which were removed from the server list earlier, so won't be processed
                total_mean_times_count += (_localMean * _localCount);
                total_count = total_count + _localCount;

                _localAggregate_average = (total_mean_times_count/total_count);
                _localAggregate_count = total_count;

                Console.WriteLine("local aggregate average = {0}", _localAggregate_average);

                System.Threading.Thread.Sleep(10000);
                foreach (var serverAddress in serverAddresses)
                {
                    using (var wscli = new ClientWebSocket())
                    {
                        var tokSrc = new CancellationTokenSource();
                        using (var task = wscli.ConnectAsync(new Uri(serverAddress.Value), tokSrc.Token))
                        {
                            task.Wait();
                        }

                        using (var task = wscli.SendAsync(new ArraySegment<byte>(Encoding.UTF8.GetBytes(thisNodeName+"~"+_localMean + "~"+_localCount)),
                            WebSocketMessageType.Text,
                            false,
                            tokSrc.Token
                            ))
                        {
                            task.Wait();
                        }
                    }

                }
            }
        }



    }
}

不要忘记通过在给定时间同步来添加静态锁或单独的事件。 (为简单起见未显示)