要求
- 巨大的 csv 文件，大小数百 GB
- 每一行包含几个数字
- 程序每次运行必须提取几千行
- 程序多次使用同一个文件，只应提取不同的行
由于 csv 文件中的行具有可变长度，因此您必须读取整个文件才能获取所需行的数据。整个文件的顺序读取仍然会非常慢 - 即使您尽可能优化文件读取。一个好的指标实际上是 wc -l 的运行时间，正如问题中的 OP 已经提到的那样。
相反，应该在算法层面进行优化。需要对数据进行一次性预处理，这样就可以快速访问某些行，而无需读取整个文件。
有几种可能的方式，例如：
- 使用带有索引的数据库
- 以编程方式创建索引文件（将行号与文件偏移量关联）
- 将csv文件转换成固定格式的二进制文件
OP 测试表明方法 1) 导致 1.5 TB 指数。方法2），创建一个连接行号和文件偏移量的小程序当然也是可以的。最后，方法 3 将允许计算文件偏移量到行号，而不需要单独的索引文件。如果每行的最大数字数已知，则此方法特别有用。否则，方法 2 和方法 3 非常相似。
下面将更详细地解释方法 3。可能还有其他要求需要稍微修改方法，但以下内容应该可以开始。
需要一次性预处理。文本 csv 行被转换为 int 数组，并使用固定的记录格式将 int 以二进制格式存储在单独的文件中。然后读取特定行 n，您可以简单地计算文件偏移量，例如与line_nr * (sizeof(int) * MAX_NUMBERS_PER_LINE);。最后，用fseeko(fp, offset, SEEK_SET); 跳转到这个偏移量并读取 MAX_NUMBERS_PER_LINE 个整数。所以你只需要读取你真正想要处理的数据。
这不仅具有程序运行速度更快的优点，而且还需要很少的主内存。
测试用例
创建了一个包含 3,000,000,000 行的测试文件。每行最多包含 10 个随机整数，以逗号分隔。
在这种情况下，这给出了一个包含大约 342 GB 数据的 csv 文件。
快速测试
time wc -l numbers.csv
给予
187.14s user 74.55s system 96% cpu 4:31.48 total
这意味着如果使用顺序文件读取方法，总共需要至少 4.5 分钟。
对于一次性预处理，转换器程序读取每一行并每行存储 10 个二进制整数。转换后的文件称为“numbers_bin”。访问 10,000 行随机选择的数据的快速测试：
time demo numbers_bin
给予
0.03s user 0.20s system 5% cpu 4.105 total
因此，此特定示例数据需要 4.1 秒，而不是 4.5 分钟。这比原来快 65 倍。
源代码
这种方法听起来可能比实际上要复杂。
让我们从转换器程序开始。它读取 csv 文件并创建一个二进制固定格式文件。
有趣的部分发生在函数 pre_process 中：在循环中使用“getline”读取一行，使用“strtok”和“strtol”提取数字并放入初始化为 0 的 int 数组中。最后这个数组使用 'fwrite' 写入输出文件。
转换过程中的错误会导致 stderr 上出现一条消息并且程序被终止。
convert.c
#include "data.h"
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <limits.h>
static void pre_process(FILE *in, FILE *out) {
int *block = get_buffer();
char *line = NULL;
size_t line_capp = 0;
while (getline(&line, &line_capp, in) > 0) {
line[strcspn(line, "\n")] = '\0';
memset(block, 0, sizeof(int) * MAX_ELEMENTS_PER_LINE);
char *token;
char *ptr = line;
int i = 0;
while ((token = strtok(ptr, ", ")) != NULL) {
if (i >= MAX_ELEMENTS_PER_LINE) {
fprintf(stderr, "too many elements in line");
exit(EXIT_FAILURE);
}
char *end_ptr;
errno = 0;
long val = strtol(token, &end_ptr, 10);
if (val > INT_MAX || val < INT_MIN || errno || *end_ptr != '\0' || end_ptr == token) {
fprintf(stderr, "value error with '%s'\n", token);
exit(EXIT_FAILURE);
}
ptr = NULL;
block[i] = (int) val;
i++;
}
fwrite(block, sizeof(int), MAX_ELEMENTS_PER_LINE, out);
}
free(block);
free(line);
}
static void one_off_pre_processing(const char *csv_in, const char *bin_out) {
FILE *in = get_file(csv_in, "rb");
FILE *out = get_file(bin_out, "wb");
pre_process(in, out);
fclose(in);
fclose(out);
}
int main(int argc, char *argv[]) {
if (argc != 3) {
fprintf(stderr, "usage: convert <in> <out>\n");
exit(EXIT_FAILURE);
}
one_off_pre_processing(argv[1], argv[2]);
return EXIT_SUCCESS;
}
数据.h
使用了一些辅助功能。它们或多或少是不言自明的。
#ifndef DATA_H
#define DATA_H
#include <stdio.h>
#include <stdint.h>
#define NUM_LINES 3000000000LL
#define MAX_ELEMENTS_PER_LINE 10
void read_data(FILE *fp, uint64_t line_nr, int *block);
FILE *get_file(const char *const file_name, char *mode);
int *get_buffer();
#endif //DATA_H
数据.c
#include "data.h"
#include <stdlib.h>
void read_data(FILE *fp, uint64_t line_nr, int *block) {
off_t offset = line_nr * (sizeof(int) * MAX_ELEMENTS_PER_LINE);
fseeko(fp, offset, SEEK_SET);
if(fread(block, sizeof(int), MAX_ELEMENTS_PER_LINE, fp) != MAX_ELEMENTS_PER_LINE) {
fprintf(stderr, "data read error for line %lld", line_nr);
exit(EXIT_FAILURE);
}
}
FILE *get_file(const char *const file_name, char *mode) {
FILE *fp;
if ((fp = fopen(file_name, mode)) == NULL) {
perror(file_name);
exit(EXIT_FAILURE);
}
return fp;
}
int *get_buffer() {
int *block = malloc(sizeof(int) * MAX_ELEMENTS_PER_LINE);
if(block == NULL) {
perror("malloc failed");
exit(EXIT_FAILURE);
}
return block;
}
demo.c
最后是一个演示程序，它读取 10,000 条随机确定的行的数据。
函数 request_lines 确定 10,000 条随机行。这些行使用 qsort 排序。读取这些行的数据。一些代码行被注释掉了。如果将它们注释掉，则读取的数据将输出到调试控制台。
#include "data.h"
#include <stdlib.h>
#include <assert.h>
#include <sys/stat.h>
static int comp(const void *lhs, const void *rhs) {
uint64_t l = *((uint64_t *) lhs);
uint64_t r = *((uint64_t *) rhs);
if (l > r) return 1;
if (l < r) return -1;
return 0;
}
static uint64_t *request_lines(uint64_t num_lines, int num_request_lines) {
assert(num_lines < UINT32_MAX);
uint64_t *request_lines = malloc(sizeof(*request_lines) * num_request_lines);
for (int i = 0; i < num_request_lines; i++) {
request_lines[i] = arc4random_uniform(num_lines);
}
qsort(request_lines, num_request_lines, sizeof(*request_lines), comp);
return request_lines;
}
#define REQUEST_LINES 10000
int main(int argc, char *argv[]) {
if (argc != 2) {
fprintf(stderr, "usage: demo <file>\n");
exit(EXIT_FAILURE);
}
struct stat stat_buf;
if (stat(argv[1], &stat_buf) == -1) {
perror(argv[1]);
exit(EXIT_FAILURE);
}
uint64_t num_lines = stat_buf.st_size / (MAX_ELEMENTS_PER_LINE * sizeof(int));
FILE *bin = get_file(argv[1], "rb");
int *block = get_buffer();
uint64_t *requests = request_lines(num_lines, REQUEST_LINES);
for (int i = 0; i < REQUEST_LINES; i++) {
read_data(bin, requests[i], block);
//do sth with the data,
//uncomment the following lines to output the data to the console
// printf("%llu: ", requests[i]);
// for (int x = 0; x < MAX_ELEMENTS_PER_LINE; x++) {
// printf("'%d' ", block[x]);
// }
// printf("\n");
}
free(requests);
free(block);
fclose(bin);
return EXIT_SUCCESS;
}
总结
这种方法提供的结果比顺序读取整个文件要快得多（样本数据每次运行 4 秒而不是 4.5 分钟）。它还需要很少的主内存。
先决条件是将数据一次性预处理为二进制格式。这种转换非常耗时，但之后使用查询程序可以非常快速地读取某些行的数据。