งูสีมิ้นต์ต้นฉบับกำลังนำลูกปัดจากรางเรียงสามสายเข้าสู่รางกลางบนโต๊ะสว่าง
pythonheapqstreamingperformance

Python heapq.merge: รวม stream ที่เรียงแล้วโดยไม่กองทั้งหมดใน RAM

พี่โชว์Engineering

การรวม log หลาย shard มักจบด้วยการต่อ list แล้ว sorted ซึ่งง่ายแต่ต้องถือข้อมูลทั้งหมดไว้พร้อมกันครับ หากแต่ละแหล่งเรียงตามเวลาอยู่แล้ว heapq.merge สร้าง iterator ที่ค่อย ๆ เลือกค่าน้อยสุดจากหัวแต่ละ stream จึงอ่านเท่าที่ consumer ต้องการ

WHY: ใช้เงื่อนไข sorted input แลกกับ memory ต่ำ

heapq.merge คล้ายเปิดสายพานหลายเส้นแล้วดูของชิ้นแรกของแต่ละเส้น เราหยิบชิ้นที่มาก่อน จากนั้นจึงเลื่อนเฉพาะสายพานนั้น ไม่จำเป็นต้องเทของทุกกล่องลงพื้นแล้วเรียงใหม่

ถ้ามี k streams และอ่านรวม n รายการ การเลือกด้วย heap ใช้เวลาประมาณ O(n log k) และเก็บหัว stream ราว O(k) ไม่นับข้อมูลที่ iterator ต้นทางถืออยู่ เทียบกับ sorted ที่มักต้อง materialize O(n) แต่ไม่ควรสรุปว่า merge เร็วกว่าเสมอ เงื่อนไขสำคัญคือทุก input ต้องเรียงด้วย key และทิศเดียวกัน

HOW: ทดสอบทั้ง key และความ lazy

import unittest
from heapq import merge
from itertools import islice


def merge_events(*streams):
    return merge(*streams, key=lambda event: event["at"])


class MergeTests(unittest.TestCase):
    def test_merges_sorted_streams(self):
        left = [{"at": 1, "id": "a"}, {"at": 4, "id": "d"}]
        right = [{"at": 2, "id": "b"}, {"at": 3, "id": "c"}]
        ids = [event["id"] for event in merge_events(left, right)]
        self.assertEqual(ids, ["a", "b", "c", "d"])

    def test_consumer_can_stop_early(self):
        values = merge(range(0, 100, 2), range(1, 100, 2))
        self.assertEqual(list(islice(values, 5)), [0, 1, 2, 3, 4])

    def test_reverse_requires_descending_inputs(self):
        result = list(merge([9, 5, 1], [8, 4, 0], reverse=True))
        self.assertEqual(result, [9, 8, 5, 4, 1, 0])


if __name__ == "__main__":
    unittest.main()

key function ใช้ตอนเปรียบเทียบ แต่ผลลัพธ์ยังเป็น object เดิม หากลำดับเท่ากันต้องตัดสินว่าความ stable ระหว่าง streams เพียงพอกับ domain หรือควรเพิ่ม sequence number ตั้งแต่ producer

เลือก merge หรือ sorted

ใช้ merge เมื่อ input ทุกชุดเรียงแล้ว ข้อมูลใหญ่หรือเป็น stream และ consumer อาจหยุดก่อนจบ ใช้ sorted เมื่อ input ยังไม่เรียง ต้องสุ่มเข้าถึงผลทั้งหมด หรือ n เล็กจนความเรียบง่ายสำคัญกว่า หากมี stream ใดผิดลำดับ merge จะไม่แจ้ง error แต่ผลจะผิดเงื่อนไข จึงควร validate ที่ producer boundary หรือมี monotonicity assertion ใน ingestion

กรณีไฟล์ให้เปิดด้วย context manager ที่มีอายุครอบการ consume iterator ไม่คืน merge iterator หลังไฟล์ถูกปิด และต้องออกแบบ backpressure เพราะ producer ที่อ่าน network อาจมี timeout ระหว่าง consumer ช้า

Recap

  • heapq.merge เป็น lazy iterator สำหรับ input ที่เรียงแล้ว
  • memory หลักสัมพันธ์กับจำนวน streams ไม่ใช่จำนวนข้อมูลรวม
  • reverse=True ต้องรับ input ที่เรียงจากมากไปน้อยอยู่แล้ว
  • ใช้ sorted เมื่อยังต้องเรียง input หรืออยากได้ list โดยตรง
  • ตรวจ monotonicity ก่อนเชื่อข้อมูลจากภายนอก

เริ่มทดลองกับ log สอง shard แล้ววัด peak memory เทียบกับ concatenate-and-sort บนขนาดจริงครับ