并行化数据处理

Question

我正在尝试改进我正在执行的某些数据处理的运行时间。数据以各种集合开始（主要是Dictionary，但还有一些其他IEnumerable 类型），处理的最终结果应该是Dictionary<DataType, List<DataPoint>>。

我的所有这些工作都很好......除了它需要将近一个小时才能运行，而且它需要在 20 分钟内运行。没有任何数据与同一集合中的任何其他数据有任何联系，尽管它们经常交叉引用其他集合，所以我想我应该并行化它。

处理的主要结构有两级循环，中间有一些处理：

// Custom class, 0.01%
var primaryData= GETPRIMARY().ToDictionary(x => x.ID, x => x);

// Custom class, 11.30%
var data1 = GETDATAONE().GroupBy(x => x.Category)
                        .ToDictionary(x => x.Key, x => x);  

// DataRows, 8.19%
var data2 = GETDATATWO().GroupBy(x => x.Type)
                        .ToDictionary(x => x.Key, x => x.OrderBy(y => y.ID));

foreach (var key in listOfKeys)
{
   // 0.01%
   var subData1 = data1[key].ToDictionary(x => x.ID, x => x);

   // 1.99%
   var subData2 = data2.GroupBy(x => x.ID)
                       .Where(x => primaryData.ContainsKey(x.Type))
                       .ToDictionary(x => x.Key, x => ProcessDataTwo(x, primaryData[x.Key]));

   // 0.70%
   var grouped = primaryData.Select(x => new { ID = x.Key, 
                                               Data1 = subData1[x.Key],
                                               Data2 = subData2[x.Key] }).ToList();
   foreach (var item in grouped)
   {
       // 62.12%
       item.Data1.Results = new Results(item.ID, item.Data2);
       // 12.37%
       item.Data1.Status = new Status(item.ID, item.Data2);
   }
   results.Add(key, grouped);
}
return results;

listOfKeys 很小，但每个grouped 将有几千个项目。 我应该如何构建这个结构，以便对item.Data1.Process(item.Data2) 的每个调用都可以排队并并行执行？

根据我的分析器，所有ToDictionary() 调用一起占用大约 21% 的时间，ToList() 占用 0.7%，内部 foreach 内的两个项目一起占用 74%。因此，我将优化重点放在了那里。

我不知道我是否应该使用Parallel.ForEach() 来替换外部foreach，内部一个，两者，或者是否应该使用其他一些结构。我也不确定我是否可以对数据（或保存它的结构）做些什么来改进对它的并行访问。

（请注意，我被困在 .NET4 上，所以无法访问 async 或 await）

Answer 1

根据您发布的百分比，您说grouped 非常大，您肯定会通过仅使内部循环瘫痪而受益。

做起来很简单

   var grouped = primaryData.Select(x => new { ID = x.Key, 
                                               Data1 = subData1[x.Key],
                                               Data2 = subData2[x.Key] }).ToList();
   Parallel.ForEach(grouped, (item) => 
   {
       item.Data1.Results = new Results(item.ID, item.Data2);
       item.Data1.Status = new Status(item.ID, item.Data2);
   });

   results.Add(key, grouped);

这假设 new Results(item.ID, item.Data2); 和 new Status(item.ID, item.Data2); 可以安全地同时进行多次初始化（我唯一担心的是它们是否在内部访问非线程安全的 static 资源，即使这样也是非线程安全的构造函数是一个真的糟糕的设计缺陷）

---

有一个很大的警告：这只会在您受 CPU 限制的情况下有所帮助。如果Results 或Status 受IO 限制（例如，它正在等待数据库调用或硬盘驱动器上的文件），这样做会伤害您的性能，而不是帮助它。如果您受 IO 限制而不是 CPU 限制，则唯一的选择是购买更快的硬件，尝试更多地优化这两种方法，或者尽可能使用内存中的缓存，这样您就不需要执行缓慢的 IO。

Answer 2

编辑

鉴于我写完这个答案后提供的时间测量，看来这种方法是在错误的地方寻找节省。我会留下我的答案作为对未经测量的优化的警告！！！

---

因此，由于您的方法的嵌套性，您正在对某些集合造成一些不必要的过度迭代，从而导致相当讨厌的 Big O 特征。

这可以通过使用 ILookup 接口通过键预先分组集合并使用这些而不是重复且昂贵的 Where 子句来缓解。

我尝试重新构想您的代码以降低复杂性（但它有点抽象）：

var data2Lookup = data2.ToLookup(x => x.Type);
var tmp1 = 
    listOfKeys
        .Select(key => 
            new {
                key, 
                subData1 = data1[key], 
                subData2 = data2Lookup[key].GroupBy(x=>x.Category)
            })
        .Select(x => 
            new{
                x.key, 
                x.subData1, 
                x.subData2, 
                subData2Lookup = x.subData2.ToLookup(y => y.Key)});
var tmp2 = 
    tmp1
        .Select(x => 
            new{
                x.key, 
                grouped = x.subData1
                            .Select(sd1 => 
                                new{
                                    Data1 = sd1, 
                                    Data2 = subData2Lookup[sd1]
                                })
            });
var result =
    tmp2
        .ToDictionary(x => x.key, x => x.grouped);

在我看来，处理在results的中途有点随意，但应该不会影响它吧？

所以一旦results 构建完成，让我们处理它...

var items = result.SelectMany(kvp => kvp.Value);
for(var item in items)
{
    item.Data1.Process(item.Data2);
}

编辑

我刻意避免使用并行 fttb，所以如果你能做到这一点，通过添加一些并行魔法可能会进一步加快速度。