如何通过并行处理数据库结果来提高性能？

Posted 2023-05-09

【中文标题】如何通过并行处理数据库结果来提高性能？

【英文标题】：How to improve performance by processing database results in parallel?

【发布时间】：2010-07-06 06:11:22

【问题描述】：

我有一个 .net 应用程序，它在 20 到 30 个 SQL 查询的区域中运行，并一次处理 1 个结果。我一直在尝试通过并行执行一些工作来提高性能。

其中 2 个查询占用了 75% 的时间，这纯粹是因为它们返回的数据量。我最初的实验是尝试使用 ntile 将这些查询分成 4 个存储桶，并并行处理每个数据读取器。如果这需要更长的时间，我认为是因为使用 NTILE + 查询数据库 4 次而不是 1 次所涉及的额外工作。

任何人都可以建议尝试其他技术还是我只是在这里浪费时间？下面的代码是实用程序类的一部分，它允许我将处理阅读器的函数排队。因此，使用我的 NTILE 实验，我将 4 个任务排队，每个任务处理 1/4 的数据（其中 ntile =1、2、3、4）并调用 Execute 以并行运行它们。

foreach (var keyValuePair in m_Tasks)
            
                var sql = keyValuePair.Key;
                var task = keyValuePair.Value;

                var conn = new OracleConnection(ConnectionString);
                conn.BeginOpen(o=> 
                    conn.EndOpen(o);
                    var cmd = conn.CreateCommand();
                    cmd.CommandText = sql;

                    cmd.BeginExecuteReader(a =>
                    
                        var reader = cmd.EndExecuteReader(a);
                        DateTime endIO = DateTime.Now;
                        Console.WriteLine(TaskName + " " + Thread.CurrentThread.ManagedThreadId + "  IO took: " + (endIO - startTime) + " ended at " + endIO);

                        DateTime taskStart = DateTime.Now;
                        task(reader);
                        DateTime endTAsk = DateTime.Now;
                        Console.WriteLine(TaskName + " " + Thread.CurrentThread.ManagedThreadId + " TAsk took: " + (endTAsk - taskStart) + " ended at " + endTAsk);
                        reader.Close();
                        conn.Close();

                        if (Interlocked.Decrement(ref numTasks) == 0)
                        
                            finishedEvent.Set();
                        

                    , null);

                ,
                null

                    );


            

            finishedEvent.WaitOne();
            DateTime endExecute = DateTime.Now;
            Console.WriteLine(TaskName + " " + Thread.CurrentThread.ManagedThreadId + " EXECUTE took: " + (endExecute - startTime) + " ended at " + endExecute);

        

感谢您的帮助。

【问题讨论】：

而不是将阅读器传递给您的任务，在读取数据之前不会推进阅读器，您可能希望先将所有数据读入内存（如果可能），然后使用多个线程之后处理数据。还要记住，正在完成的工作应该超过启动新线程的时间。如果您可以独立处理每条记录，那么您可以在读取每条记录后将其传递给工作线程，从而防止您的阅读器阻塞。

嗨，米凯尔。您的意思是使用数据集然后排队作业来处理每条记录吗？我正在处理大量数据，所以如果可能的话，我宁愿使用 DataReader。

Mikael，如果他的问题是 I/O 限制的，那不是只会让事情变得更糟吗？

如果我的问题是 I/O 绑定问题，是否有任何线程帮助或我只是在浪费时间？

Neil - 以下任何想法（ROWID 上的 NTILE，MOD Id 上的功能索引）是否对您有用 - 如果是这样，我很高兴将其写成更正式的答案而不是评论。

【参考方案1】：

我认为你是对的，执行 NTILE 的成本超过了节省并行性。

您需要使用将查询集拆分为明确分离的集合的东西。

如果您的查询返回的数据少于（大约）总数据的 15%，那么分解索引（索引字段或功能索引）上的表可能是您最好的起点。

示例： 假设您的数据在每一行上都有一个数字伪键，请在 MOD(Id,4) 上创建一个功能索引 - 这将为您提供基于索引的 NTILE 方法版本。 （我认为您不能在 NTILE 上拥有功能索引）。

这种特定方法可能会适得其反——您会从不同线程中的相同块获取数据，因此可能会增加 I/O（取决于内存）。

Oracle 并行查询倾向于执行此操作的方式（假设您要处理表中超过 15% 的数据）是简单地将表分成 N 个物理块（使用 rowid），然后运行 ​​N'full扫描这些块。

我不确定您是否可以从前端复制这种方法。对 key id 进行拆分会增加通过索引到每一行的成本。

您可能想要的是通过键以外的其他方式拆分表，或者如果您按键拆分，则按范围而不是 NTILE 方法拆分它。

【讨论】：

您不能在NTILE 上拥有功能索引，因为它不是确定性的。但是，您可以NTILE(n) over (order by ROWID)。这将在功能上将数据集分解为离散的物理块。我怀疑 OP 应该得到一个 SQL 跟踪，等待查看添加滚动你自己的并行性是否已经使 I/O 子系统饱和。

做了一个快速测试，看起来像在 ROWID 上执行 NTILE（然后通过 NTILE 结果限制查询）仍会在每个线程上触发全表扫描 - 它看起来不像做任何“魔术”来快速将整个集合分成偶数块。在我的测试系统上，按主键的 NTILE 的成本略低，但里程可能会有所不同。 MOD(Id,10) 上的功能索引对我来说便宜很多；我认为可以通过使用一个函数来实现更好的结果，该函数可以使相同块中的数据在相同“块”中得到更好的比例，但这可能是一个好的开始。

【参考方案2】：

我使用 OracleCommand.Fetchsize 来提高大型查询的性能。

cmd.FetchSize = &H100000  '1Mb
Dim Rdr = cmd.ExecuteReader

前段时间，我使用异步读取器来获取 Blob 数据。但是要使用异步读取器，您需要维护一个数组，每个异步结果都有一个循环，直到最后一个读取器结束。

   Public Shared Function FromBlob(ByVal Id As String, ByVal Rv As String, ByVal cn As OracleConnection) As Proyecto
     Dim n As Integer, Prj As Proyecto = Nothing
     Dim Bf(2)() As Byte, arrAr(2) As IAsyncResult 'Para proceso asíncrono

     Dim Cmd As New OracleCommand( _
         "Select rv,fecha,Datos From Proyectos Where Id=:Id and Rv in (:Rv,'Av','Est')", cn)
     Cmd.BindByName = True
     Cmd.Parameters.Add("Id", OracleDbType.Varchar2, Id, ParameterDirection.Input)
     Cmd.Parameters.Add("Rv", OracleDbType.Varchar2, Rv, ParameterDirection.Input)
     If Rv Is Nothing Then Prj = Proyecto.Actprj
     Try
        Using Rdr As OracleDataReader = Cmd.ExecuteReader
            Do Until Rdr.Read = False
                Dim rv1 As String = Rdr.GetString(0)
                Select Case rv1
                    Case "Av" : n = 1   'Avance TND
                    Case "Est" : n = 2  'Datos Seguimiento Estudio Seguridad
                    Case Else : n = 0
                End Select
                If Rdr.IsDBNull(2) = False Then
                   Dim Blob As OracleBlob = Rdr.GetOracleBlob(2)
                   Dim Buffer(CInt(Blob.Length)) As Byte
                   Bf(n) = Buffer
                   arrAr(n) = Blob.BeginRead(Buffer, 0, Buffer.Length, Nothing, Blob)
                End If
            Loop
            If Bf(0) Is Nothing AndAlso Prj Is Nothing Then _
               MessageBox.Show("Fallo al cargar proyecto") : Return Nothing
            For n = 0 To Bf.Length - 1
                Dim ar As IAsyncResult = arrAr(n)
                If ar IsNot Nothing AndAlso ar.AsyncWaitHandle.WaitOne() Then
                   Dim blob As OracleBlob = DirectCast(ar.AsyncState, OracleBlob)
                   blob.EndRead(ar)
                   blob.Dispose()
                   If ar.IsCompleted Then
                      Using rd As New BinReader(New MemoryStream(Bf(n)))
                          If n = 0 Then
                             Prj = New Proyecto(rd, False)
                          Else
                             Dim entry = Proyecto.Entry.FromLob(rd), Index = Prj.IndexOf(entry)
                             If Index < 0 Then Prj.Add(entry) Else Prj(Index) = entry
                          End If
                      End Using
                   End If
                End If
            Next
        End Using
        Catch ex As Exception
            MessageBox.Show(ex.Message)
     End Try
     Return Prj
  End Function

【讨论】：

注意：当我使用 BeginRead (Async) 时，我不会在我的应用程序上使用 multiTask，Oracle 在分配的缓冲区 (Bf) 上执行同步读取器。我只需要验证每个 IAsyncResult 即可完成任务。在三个阅读器的代码示例中，结果大约快 50%，但这是为了减少往返，而不是为了数据大小。

【参考方案3】：

您可以将 Ref Cursor 与 Oracle 一起使用，通过一个 OracleCommand 执行一些 Sql：

  Dim cmd As New OracleCommand("Begin " _
  & "Open :1 for Select T.CODTRA,SIM,JLA CAL,SUP,RESP,SERV,SubStr(Aparato,1,3) SIS,PERS,(nvl(DUR,0) * 60) as Dur,t.DESTRA,g.DesTra Destrae,OBS from " & TraRec & " T, Trarec_Gee g where T.codtra <> 'RV' and T.Codtra=G.Codtra(+);" _
  & "Open :2 for Select Red,descr from Redes;" _
  & "Open :3 for Select * from Tr_Redes;" _
  & "Open :4 for Select CODTRA,T_COND,COND,DEMORA * 60 as DEMORA from " & TrCondic _
  & ";end;", cn)

  For n = 0 To 3 : cmd.Parameters.Add(Nothing, OracleDbType.RefCursor, ParameterDirection.Output) : Next
  Dim da As New OracleDataAdapter(cmd)
  da.Fill(0, 0, ds.Tnd, ds.Redes, ds.TrRedes, ds.TrCondic)

注意：Da.Fill(0, 0, T1, T2 ...) 是 Oracle 特有的函数，用于在单个语句中检索多个表。

【参考方案4】：

最终证明这是一个 IO 绑定问题。我已经能够通过异步执行 IO 来实现性能改进。 ROWID 上的 NTILE 可以满足我的要求，但到目前为止它没有帮助，因为问题是 IO 绑定的。