循环遍历大量行

Question

循环遍历大量行

3

我在循环处理包含100万个潜在行的数据库时遇到了时间问题。我将行数据提取到一个DataTable中并进行循环操作，但是这样做已经变得很慢了。还有其他的方法吗？我可以将这些行拆分为每个20000个记录一组。可以在C#中使用并行处理吗？基本上代码循环遍历与特定查询匹配的每个潜在记录，并尝试确定它是否是合法条目。这就是为什么需要逐个访问每个记录的原因。每个对象的一条记录可能达到1000万行。解决方案似乎是多台计算机上的并行处理或单台机器上的PP和多核心处理，或者是某种数据结构/方法的改变？

任何意见、想法和猜测都有助于使这个过程更快捷和合理。

- iefpw

2个回答

0

我建议使用双核机器进行并行循环，并尝试使用通用列表的 for each 循环，我认为这可能会使您的处理速度更快。

- Alex

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- ntziolis · Accepted Answer

首先，不要使用 DataTable 进行此类操作：

它很慢
它消耗的内存太多
你需要等待很长时间才能开始处理数据
- 在此期间，其他核心什么都没做，因为将数据读入 DataTable 无法并行化。
- 同时，在读取数据时，CPU 通常几乎未被利用，因为网络或其他 I/O 延迟通常是主要因素。

所以再次强调：不要使用 DataTable 进行此类操作。

相反，使用 DataReader。这允许您立即开始消费/处理数据，而不必等待其加载。最简单的版本如下（适用于 MS SQL Server 的示例）：

var command = new SqlCommand()
{
  CommandText = "SELECT * FROM Table";
  Connection = new SqlConnection("InsertConnectionString");
};

using(var reader = command.ExecuteReader())
{
  while(reader.Read())
  {
    var values = new object[reader.FieldCount];
    reader.GetValues(values);

    // process values of row
  }
}

读者在执行处理代码时会被阻塞，这意味着不再从数据库中读取更多行。如果处理代码很重，使用Task库创建执行检查的任务可能是值得的，这将使您能够利用多个核心。但是，创建Task会有一些开销，如果一个Task不包含足够的“工作”，您可以将几行批处理在一起。

public void ReadData()
{
  var taskList = new List<Task<SomeResultType>>();

  var command = new SqlCommand()
  {
    CommandText = "SELECT * FROM Table";
    Connection = new SqlConnection("InsertConnectionString");
  };
  using(var reader = command.ExecuteReader())
  {
    var valueList = new List<object[]>(100);
    while(reader.Read())
    {
      var values = new object[reader.FieldCount];
      reader.GetValues(values);

      valueList.Add(values);

      if(valueList.Count == 100)
      {
        var localValueList = valueList.ToList();
        valueList.Clear();

        taskList.Add(Task<SomeResultType>.Factory.StartNew(() => Process(localValueList));
      }
    }
    if(valueList.Count > 0)
      taskList.Add(Task<SomeResultType>.Factory.StartNew(() => Process(valueList));
  }

  // this line completes when all tasks are done
  Task.WaitAll(taskList.ToArray());
}

public SomeResultType Process(List<object[]> valueList)
{
  foreach(var vals in valueList)
  {
    // put your processing code here, be sure to synchronize your actions properly
  }  
}

批处理大小（当前为100）取决于实际处理情况，可能需要进行调整。
同步会带来自己的挑战，您需要非常小心地处理共享资源。