如何加速PowerShell中从DataSet向SQL批量插入数据？

运行以下PowerShell脚本从DataSet向SQL表逐行插入数据时，耗时极长，插入时间随数据行数增加而增长。请问是否有方法加速该过程，或拆分数据实现并行插入？

for ( $i = 0; $i -lt $DataSet.Tables[0].Rows.Count; $i++) { 
  try { 
    $valuestr = New-Object -TypeName System.Text.StringBuilder 
    for ( $x = 0; $x -lt 11; $x++) { 
      if ($x -lt 10) { 
        [void]$valuestr.Append("'" + $DataSet.Tables[0].Rows[$i][$x].ToString().Trim().Replace("'", "/") + "',") 
      } 
      else { 
        [void]$valuestr.Append("'" + $DataSet.Tables[0].Rows[$i][$x].ToString().Trim().Replace("'", "/") + "'") 
      } 
    }
    [string]$inputstr = $valuestr.ToString()
    [char[]]$values = $inputstr.ToCharArray()
    [string]$output = ''
    foreach ($letter in $values) { 
            
      [int]$value = [Convert]::ToInt32($letter)
      [string]$hexOutput = [String]::Format("{0:X}", $value)
      switch ($hexOutput) {
        "627" { $output += "ا"; Break }
        default { $output += [Convert]::Tostring($letter); break }
      }
    }
    $sqlCmdd.CommandText = "INSERT INTO $sqlTable ([ICN],[ICN_NODE],[PARTITION_ID],[ICN_CREATE_DT] ,[ICN_SEQ_NUM],[ICN_SUBNUM],[ICN_TAG_ID],[ICN_TAG_VER],[BLK_NUM],[BLK_LEN] ,[FREE_FORM_TXT]) Values (" + $output.ToString() + ")" 
    $sqlCmdd.ExecuteNonQuery() 
  }
  catch { $_.Exception.Message | Out-File C:\log\log.txt -Append } 
}

1. 使用SqlBulkCopy（最推荐的高效方案） #

这是SQL Server官方提供的批量写入工具，性能远高于逐行插入，能将数据写入效率提升一个数量级以上。核心思路是直接将预处理后的DataTable批量写入SQL表：

# 先预处理DataTable中的数据（替换特殊字符、转换目标字符）
foreach ($row in $DataSet.Tables[0].Rows) {
    for ($x = 0; $x -lt 11; $x++) {
        $rawValue = $row[$x].ToString().Trim()
        # 替换单引号避免SQL语法错误
        $processedValue = $rawValue.Replace("'", "/")
        # 处理字符转换逻辑
        $charArray = $processedValue.ToCharArray()
        $finalValue = ''
        foreach ($letter in $charArray) {
            $hex = [String]::Format("{0:X}", [Convert]::ToInt32($letter))
            $finalValue += if ($hex -eq "627") { "ا" } else { $letter.ToString() }
        }
        $row[$x] = $finalValue
    }
}

# 初始化SqlBulkCopy并执行批量插入
$connectionString = "你的数据库连接字符串"
$bulkCopy = New-Object Data.SqlClient.SqlBulkCopy($connectionString)
$bulkCopy.DestinationTableName = $sqlTable

# 自动映射列（如果DataSet列名与SQL表列名完全一致可省略）
foreach ($col in $DataSet.Tables[0].Columns) {
    $bulkCopy.ColumnMappings.Add($col.ColumnName, $col.ColumnName) | Out-Null
}

try {
    $bulkCopy.WriteToServer($DataSet.Tables[0])
} catch {
    $_.Exception.Message | Out-File C:\log\log.txt -Append
} finally {
    $bulkCopy.Close()
}

2. 参数化批量INSERT（兼容场景备选） #

如果无法使用SqlBulkCopy，可将多行数据合并为一个INSERT语句（比如一次插入100-1000行），配合参数化查询减少网络往返次数，同时避免SQL注入风险：

$batchSize = 100
$rowsCount = $DataSet.Tables[0].Rows.Count
$connection = New-Object Data.SqlClient.SqlConnection($connectionString)
$connection.Open()
$transaction = $connection.BeginTransaction()

try {
    for ($batchStart = 0; $batchStart -lt $rowsCount; $batchStart += $batchSize) {
        $batchEnd = [math]::Min($batchStart + $batchSize - 1, $rowsCount - 1)
        $valueClauses = @()
        $params = @()

        for ($i = $batchStart; $i -le $batchEnd; $i++) {
            $row = $DataSet.Tables[0].Rows[$i]
            $paramNames = @()
            for ($x = 0; $x -lt 11; $x++) {
                $paramName = "@p_$($i)_$($x)"
                $paramNames += $paramName
                # 预处理参数值
                $rawValue = $row[$x].ToString().Trim().Replace("'", "/")
                $charArray = $rawValue.ToCharArray()
                $finalValue = ''
                foreach ($letter in $charArray) {
                    $hex = [String]::Format("{0:X}", [Convert]::ToInt32($letter))
                    $finalValue += if ($hex -eq "627") { "ا" } else { $letter.ToString() }
                }
                $params += New-Object Data.SqlClient.SqlParameter($paramName, $finalValue)
            }
            $valueClauses += "($($paramNames -join ','))"
        }

        $sql = "INSERT INTO $sqlTable ([ICN],[ICN_NODE],[PARTITION_ID],[ICN_CREATE_DT],[ICN_SEQ_NUM],[ICN_SUBNUM],[ICN_TAG_ID],[ICN_TAG_VER],[BLK_NUM],[BLK_LEN],[FREE_FORM_TXT]) VALUES $($valueClauses -join ',')"
        $cmd = New-Object Data.SqlClient.SqlCommand($sql, $connection, $transaction)
        $cmd.Parameters.AddRange($params)
        $cmd.ExecuteNonQuery()
    }
    $transaction.Commit()
} catch {
    $transaction.Rollback()
    $_.Exception.Message | Out-File C:\log\log.txt -Append
} finally {
    $connection.Close()
}

3. 并行插入（大数据集场景补充） #

针对超大规模数据集，可拆分数据为多个批次，通过RunspacePool实现并行插入，注意控制并行度避免数据库锁竞争：

$batchSize = 2000
$rowsCount = $DataSet.Tables[0].Rows.Count
$batches = [math]::Ceiling($rowsCount / $batchSize)
$connectionString = "你的数据库连接字符串"

# 创建Runspace池，控制并行度（建议4-8，根据服务器性能调整）
$runspacePool = [RunspaceFactory]::CreateRunspacePool(1, 6)
$runspacePool.Open()
$jobs = @()

for ($batchIdx = 0; $batchIdx -lt $batches; $batchIdx++) {
    $startRow = $batchIdx * $batchSize
    $endRow = [math]::Min(($batchIdx + 1) * $batchSize - 1, $rowsCount - 1)

    $scriptBlock = {
        param($start, $end, $sourceTable, $sqlTable, $connStr)
        # 创建子DataTable
        $subTable = $sourceTable.Clone()
        for ($i = $start; $i -le $end; $i++) {
            $subTable.ImportRow($sourceTable.Rows[$i])
        }
        # 预处理子表数据
        foreach ($row in $subTable.Rows) {
            for ($x = 0; $x -lt 11; $x++) {
                $rawValue = $row[$x].ToString().Trim().Replace("'", "/")
                $charArray = $rawValue.ToCharArray()
                $finalValue = ''
                foreach ($letter in $charArray) {
                    $hex = [String]::Format("{0:X}", [Convert]::ToInt32($letter))
                    $finalValue += if ($hex -eq "627") { "ا" } else { $letter.ToString() }
                }
                $row[$x] = $finalValue
            }
        }
        # 批量插入子表
        $bulkCopy = New-Object Data.SqlClient.SqlBulkCopy($connStr)
        $bulkCopy.DestinationTableName = $sqlTable
        $bulkCopy.WriteToServer($subTable)
        $bulkCopy.Close()
    }

    $job = [PowerShell]::Create().AddScript($scriptBlock).AddArgument($startRow).AddArgument($endRow).AddArgument($DataSet.Tables[0]).AddArgument($sqlTable).AddArgument($connectionString)
    $job.RunspacePool = $runspacePool
    $jobs += $job
    $null = $job.BeginInvoke()
}

# 等待所有并行任务完成
foreach ($job in $jobs) {
    $job.EndInvoke($job.BeginInvoke())
    $job.Dispose()
}
$runspacePool.Close()
$runspacePool.Dispose()

4. 原有代码的局部优化 #

如果必须保留逐行插入逻辑，可做以下优化减少耗时：

• 复用StringBuilder对象，避免循环内重复创建
• 直接处理字段值而非拼接后的字符串，减少字符遍历次数
• 开启数据库事务，减少日志写入开销

内容的提问来源于stack exchange，提问作者MustaMusta