C#大数据处理：从LINQ到并行流

轻松学习Python，无需理论负担。在浏览器中直接解决带自动检查的实践任务，无需安装任何东西。

为什么C#是您处理大数据的秘密武器？

\n\n

当谈到大数据处理时，人们首先想到的是Python及其Pandas、Scala与Spark或Java与Hadoop。但说实话：C#常常被不公平地忽视。这很可惜。.NET平台为我们提供了如此强大的工具，如PLINQ、MemoryMappedFile、Parallel.For和异步流，使我们能够在普通硬件上处理GB级数据，而无需大费周章。

\n\n

在本文中，我们不会空谈理论——我们将直接付诸实践。您将学习如何充分利用您的CPU进行C#大数据处理，如何避免OutOfMemoryException，以及如何处理数十GB大小的文件，就像处理一本大书一样简单。

\n\n

主要问题：如何避免内存耗尽？

\n\n

想象一下：您有一个50GB的CSV文件，包含服务器日志。您用普通的StreamReader打开它并逐行读取。一切顺利吗？不。如果您试图将其全部加载到List<string>中——就会遇到OutOfMemoryException。即使您有64GB RAM，.NET也可能导致内存碎片化，使得大对象无法放入堆中。

\n\n

第一个解决方案：流式处理和分块处理。不要一次性加载所有内容——要一点一点地处理。

\n\n

public async IAsyncEnumerable<string[]> ReadCsvChunksAsync(string filePath, int chunkSize = 10000)\n{\n    using var reader = new StreamReader(filePath);\n    string? line;\n    var buffer = new List<string>(chunkSize);\n\n    while ((line = await reader.ReadLineAsync()) != null)\n    {\n        buffer.Add(line);\n        if (buffer.Count >= chunkSize)\n        {\n            yield return buffer.ToArray();\n            buffer.Clear();\n        }\n    }\n\n    if (buffer.Count > 0)\n        yield return buffer.ToArray();\n}\n

\n\n

现在，您可以处理GB级数据而无需担心内存：

\n\n

await foreach (var chunk in ReadCsvChunksAsync("logs.csv"))\n{\n    ProcessChunk(chunk); // 您的逻辑\n}\n

\n\n

并行处理：充分利用所有CPU核心

\n\n

好的，我们分块读取数据。但如果您有16核CPU，而数据是顺序处理的——您只使用了其6%的性能。是时候开启C#并行计算了。

\n\n

PLINQ — Parallel LINQ

\n\n

最简单的魔法是在您的LINQ查询中添加.AsParallel()。但在处理大数据时需要小心：PLINQ自行决定何时并行化，这可能导致过度碎片化。

\n\n

var numbers = Enumerable.Range(1, 10_000_000);\n\n// 普通LINQ（慢）\nvar sum = numbers.Where(n => n % 2 == 0).Sum();\n\n// PLINQ（快，但需谨慎）\nvar parallelSum = numbers\n    .AsParallel()\n    .WithDegreeOfParallelism(Environment.ProcessorCount)\n    .Where(n => n % 2 == 0)\n    .Sum();\n\nConsole.WriteLine($"结果: {parallelSum}");\n

\n\n

但PLINQ并非万能药。对于复杂的C#大数据处理，最好使用Parallel.ForEach并手动控制分区。

\n\n

带范围划分的Parallel.ForEach

\n\n

var allLines = File.ReadLines("huge_dataset.csv");\nvar processedCount = 0;\nvar lockObject = new object();\n\nParallel.ForEach(\n    allLines,\n    new ParallelOptions { MaxDegreeOfParallelism = Environment.ProcessorCount },\n    line =>\n    {\n        var result = ProcessLine(line); // 您的繁重函数\n        lock (lockObject)\n        {\n            processedCount++;\n            if (processedCount % 100_000 == 0)\n                Console.WriteLine($"已处理: {processedCount}");\n        }\n    });\n

\n\n