海量数据处理方案

貪嗔痴

海量数据处理方案 371什么是海量数据？所谓的海量数据从字面上理解就是数据多到已经用大海来形容了，它指的就是数据量太大，无法在较短时间内迅速解决，无法一次性装入内存。少量数据处理方案对于在内存中可以一次性快速处理的少量数据，我们有很多方式对数据进行处理。数据排序：通过快排、归并排序、桶排序等等多种方式都可以进行处理；数据查询：通过遍历、二分查找等方式，或者通过构建 hash 表、红黑树、位图等数据结构实现快速查询；求 TOPK：对于静态数据，可以使用排序后遍历，或者使用快排的思想快速查找分区点等方式进行处理；对于动态数据，可以借助堆这种数据结构实现快速查找 TOPK；当然，静态数据也能使用堆的方式求 TOPK；两组数据去重/找重：对一组数据构建 hash 表，另一组数据基于这组 hash 表实现去重/找重；先对两组数据分别进行排序，然后使用双指针（分别指向一组数据）的方式进行逐个遍历，实现去重/找重；对于少量数据的处理，因为当前数据量较小，可以一次性加载到内存中进行处理，所以对于这种数据的处理，我们可以找到很多种方式对数据进行处理。海量数据处理面临的问题我们要想对海量数据实现排序、查询、求 TOPK、去重等操作，我们没法直接把数据一次性加载到内存中，然后一次性进行处理，因为海量数据往往面临以下两个问题：单台机器内存不够；单台机器对数据的处理速度过慢。海量数据处理的核心思想基于海量数据处理面临的上述两个问题，我们可以很容易想到一些对于海量数据进行处理的方案：不必把数据一次性加载到内存中，而是通过分批处理的方式，把外存中的数据加载到内存中进行处理；单机内存存不下，那么可以扩展为多机，对于外存中的海量数据，把数据分片到不同的机器中，用多机内存进行处理；对于单机对数据处理速度慢的问题，可以通过多机并行计算的方式进行并行处理，提升整体的处理速度。海量数据处理的一些常见案例及对应处理方案排序问题案例：给 10 GB 的订单文件进行排序，排序条件是订单的总金额。首先需要判断，当前内存中能否一次性处理这 10 GB 的文件？如果能处理的话，直接在内存中通过排序的方式进行处理即可。如果内存中无法一次性处理 10 GB 大小的文件，我们应该如何处理？我们假设内存大小是 1.5 GB ，我们每次从 10 GB 的文件中读取 1 GB （预留 0.5 GB 的空间，用来申请临时变量等使用）的文件到内存中；我们在内存中对这 1 GB 的文件排好序后，将排好序的 1 GB 文件写到外存中，然后释放内存空间；接着继续往下处理第 2 个 1 GB、第三个 1 GB ……依次类推，最终可以得到 10 个 1 GB 的小文件，每个小文件内部都是有序的。接下来，我们需要将这 10 个各自有序的 1 GB 小文件合并成一个大的 10 GB 的有序文件；这里的处理思路也非常简单，我们可以从每个小文件中取最小的一个值，放入内存中 size 为 10 的数组中，找到数组中最小的值，写 10 GB 的新文件（也可以是覆盖原来的 10 GB 文件）第一个位置；然后从内存中删除这个最小值，并在该值对应的 1 GB 有序小文件中读取第二个值放到数组中，重复之前的操作；最终我们即可将 10 个 1 GB 的有序文件整合成 1 个 10 GB 的有序文件。针对上面这种情况，其实我们还可以继续做一些优化：（1）IO 操作是比较耗时的，所以我们每次不仅仅只是从外存中读取 1 个值处理，可以在读取时给每个小文件添加一层缓存，大小 100 MB ，读取时优先从缓存中获取，缓存读完后再触发下一次 IO 操作从文件中读取数据；（2）同样的思路，在写回文件时，也可以使用缓存来减少 IO 操作，提高效率；（3）上述问题在单机的情况下，我们顺序将 10 GB 的文件读取为 10 个 1 GB 的小文件，然后用多路归并的方式合并；在多机情况下，假设我们有 10 台机器，那么每台机器可以读取 1 GB 的文件，同时进行排序，再按照上面的思路进行后续的整合，可以进一步提高效率。对于上述问题，我们使用了多路归并的思想进行了处理，实际情况中，如果订单金额最小最大值我们是已知的，且分布较为均匀的话，我们还可以使用如下的方式进行处理：假设订单金额在 0-999 元的范围内，顺序读取 10 GB 文件，按照金额大小，将订单拆分到不同的桶中；之后，按序将每个小文件读取到内存中排好序，然后写回到 10 GB 的大文件中；由于不同小文件之间金额有大小关系，所以顺序写回的10 个小文件写好后整个大文件就是有序的。查询问题案例：判断当前登录人 IP 地址是否在 IP 地址白名单内，已知 IP 地址白名单现有 10 亿个 IP 地址。单机处理方式：这个问题首先我们已知 IP 地址可以用 32 位的整数来表示，占用 4 个字节；对于 10 亿个 IP 地址，总共占用空间大约 4 GB；假设我们内存空间是 1 GB，那么是否可以在内存中直接判断登录人 IP 是否在这 10 亿个 IP 地址中呢？首先考虑可以支持快速查找的数据结构：hash 表、红黑树，发现这种数据结构无法在 1 GB 的内存地址存储 4 GB 的数据；考虑到节省空间，可以使用位图来进行处理：IP 地址表示为 32 位整数，int 类型数字范围最大可以在 0~2^32-1，所以存储 int 类型数字，我们总共需要 2^32 个二进制位来进行存储；8 个二进制位为1个字节，所以 2^32 / 8 个字节，可得大约需要 500 M 空间的位图来存储。位图构建：扫描 IP 地址白名单文件，找到 IP 地址对应 int 类型数字所在的位图中的二进制位的位置，将对应二进制位置为1；实时查询：通过当前登录人 IP，找到位图中对应的位置，判断当前位是否为 1 即可处理。多机处理方式：对于白名单服务，我们也可以考虑使用多机进行处理，降低单机的压力。假设我们有 4 台机器，每台机器有 1 GB 的内存，我们需要处理这 10 亿个 IP 地址（ 500 MB ）。我们首先需要扫描外存中的白名单文件，对每个 IP 地址求 MD5（确保分布均匀），再对 4 求余，然后将该条白名单 IP 存储对应的机器内存中（存储形式按照内存是否富余可以考虑 hash 表、红黑树、位图等方式）。在校验当前登录人 IP 时，我们也对当前登录人 IP 求 MD5，再对 4 求余，然后找到对应的机器，再在该机器上判断是否在白名单内即可。TOPK 问题案例：对 100 GB 搜索关键字⽂件，统计其中出现频率最高的 TOP 100 的关键词针对这类问题，我们首先假设内存足够大，可以直接一次性加载处理的情况，我们应该如何处理？（1）首先统计每个关键词出现的频率，可以用如下方法方法1：先进行排序，然后顺序扫描，边扫描边统计；方法2：使用 hash 表进行频率统计（2）然后我们可以使用堆的方式，按照频率统计出现频率 TOP 100 的关键词。简单的代码实现如下：public String[] top100(String[] keywords) {  if (keywords == null) return null;  // hash表统计频率  Map map = new HashMap();  for (String keyword : keywords) {    map.put(keyword, map.getOrDefault(keyword, 0) + 1);  }  // 小顶堆保留频率最大的top100  riorityQueue heap = new riorityQueue(100, (o1, o2) -> map.get(o1) - map.get(o2));  for (Map.Entry entry : map.entrySet()) {    String keyword = entry.getKey();    int frequency = entry.getValue();    if (heap.size()  map.get(heap.peek())) {      heap.poll();      heap.offer(keyword);    }  }  // 结果输出  String[] res = new String[100];  int i = 0;  while (!heap.isEmpty()) {    res[i++] = heap.poll();  }  return res;}实际上，100 GB 数据无法一次性加载到内存中，对于海量数据的频率 topK ，我们按如下方式进行处理：（1）首先还是需要对每个关键词的频率进行统计，统计关键词频率的方式跟内存中统计一样，都可以使用排序后遍历和 hash 表统计的方式。a. 排序后遍历的方式较为简单，首先对于海量数据排序，我们可以使用之前提到的海量数据排序问题的处理方式，得到一个有序的关键词文件；之后我们顺序扫描有序文件中的关键词到内存中，并记录同一关键字连续出现的个数，统计每个关键词的形式，之后以 key-val 的形式（{"关键字":"出现次数"}）存储到外存中形成一份关键词频率文件。b. 对于 100 GB 海量数据关键词频率的统计，还可以使用 hash 表的方式：假设机器有 15 GB 的空间，我们对每个关键词求 MD5，然后对 10 取余的方式进行数据分片，这样余数就落到 0-9 的范围内，就可以把 1 个大文件拆分为 10 个小文件，编号为 0.txt~9.txt；我们可以先把 0.txt 加载到内存中，然后使用 hash 表来进行频率统计，统计后生成 key-val 的频率统计小文件；之后依次读取 1.txt~9.txt，这样就生成 10 个存储 key-val 的频率统计小文件。（2）统计完频率以后，我们需要读取频率文件，然后获取频率最高的 top100 关键词。这里依然可以使用堆的方式，读取外存中的 key-val 到内存中，然后通过容量为 100 的小顶堆获取频率最高的 top100 关键词。去重/找重问题案例：有两个文件 0.txt 和 1.txt，各有 50 亿条 URL，每条占内存为 64 字节，单机内存限制 4 GB，需要找出这两个文件中相同的 URL我们首先考虑内存足够大，可以一次性将所有数据加载到内存中的情况，如何处理两个文件中找到重复 URL 的问题，有两种方式可以作为参考：（1）方式1：先对两个文件各自进行排序，然后使用双指针的方式，a 指针指向文件 0.txt 开头，b 指针指向文件 1.txt 开头，对比指针位置是否相等，相等就将重复值取出保存；不相等的情况下，如果 a 指针的值小于 b 指针的值，那么 a 指针后移一位，否则 b 指针后移一位，可参考如下代码：public List findDuplicate(int[] arr1, int[] arr2) {  if (arr1 == null || arr2 == null) return null;  // 先对两个数组进行排序  Arrays.sort(arr1);  Arrays.sort(arr2);  // 重复数字结果  List res = new ArrayList();  // a、b双指针  int a = 0, b = 0;  while (a