Find Median from Data Stream

Q: ما التعقيد الزمني لـ Find Median from Data Stream؟

باستخدام كومتين، تكون addNum بتعقيد O(log n) ، مع ثلاث عمليات دفع أو سحب من الكومة كحد أقصى، وتكون findMedian بتعقيد O(1) ، أي عمليتا قراءة. تُخزَّن كل قيمة، لذا تكون المساحة O(n) . أما الفرز عند كل استعلام فيكلف O(n log n) لكل عملية findMedian .

Q: لماذا لا نحتفظ بالقيم في مصفوفة مرتبة؟

تُجيب المصفوفة المرتبة عن findMedian في O(1) ، ويجد البحث الثنائي موضع القيمة الجديدة في O(log n) . لكن إدراجها يزيح كل قيمة تأتي بعدها، بتعقيد O(n) لكل addNum .

صعبالكومة

+10

صمّم بنية تدعم عمليتين: تخزّن addNum عددًا صحيحًا، وتُعيد findMedian وسيط جميع القيم المخزّنة حتى الآن. وسيط عدد فردي من القيم هو القيمة الوسطى؛ أما إذا كان عددها زوجيًا، فهو متوسط القيمتين الوسطيتين.

العملية i هي ops[i]، وتحتوي values[i] على العدد عند تنفيذ addNum وعلى 0 عند تنفيذ findMedian. نفّذ العمليات بالترتيب على بنية فارغة، وأعِد سلسلة نصية لكل عملية: "null" لـ addNum، والوسيط لـ findMedian. اكتب الوسيط مثل "3" أو "2.5" أو "-0.5": من دون فاصلة عشرية للعدد الصحيح، واكتب الصفر على شكل "0"، وليس أبدًا "-0".

الدالة

medianStreamOps(ops: string-array, values: integer-array) → string-array

opsstring-array: العمليات، بالترتيب الذي تُنفَّذ به
valuesinteger-array: العدد لكل addNum، و0 لكل findMedian
تُرجعstring-array: إجابة واحدة لكل عملية، كنص

القيود

1 ≤ ops.length ≤ 3000
values.length == ops.length
كل قيمة من ops[i] هي addNum أو findMedian.
-10⁵ ≤ values[i] ≤ 10⁵ بالنسبة إلى addNum، وvalues[i] == 0 بالنسبة إلى findMedian.
findMedian لا يتم استدعاؤها إلا بعد استدعاء addNum مرة واحدة على الأقل.

أمثلة

المدخلات: ops = ["addNum", "findMedian", "addNum", "findMedian", "addNum", "findMedian", "addNum", "findMedian"]values = [5, 0, 1, 0, 8, 0, 3, 0]
المخرجات: ["null", "5", "null", "3", "null", "5", "null", "4"]
الشرح: مع 1 و5، يكون الوسيط هو (1 + 5) / 2 = 3. ومع 1 و5 و8، يكون 5، ومع 1 و3 و5 و8، يكون (3 + 5) / 2 = 4.

سؤال إضافي

إذا كانت كل قيمة تقع بين 0 و100، فكيف يمكنك جعل العمليتين تعملان في زمن ثابت؟

تلميحات

افتحها واحدًا تلو الآخر. كل تلميح يكشف أكثر قليلًا.

فُتح 0 من 3

الحل

يقع الوسيط بين نصفي البيانات، لذا لا يهم الترتيب داخل كل نصف؛ المهم فقط أكبر قيمة في النصف الأدنى وأصغر قيمة في النصف الأعلى. تُبقي كومتان هاتين القيمتين بالضبط في القمة. تستغرق كل عملية addNum زمنًا O(log n)، وتقرأ كل عملية findMedian قيمتي القمة في O(1).

الفرز عند كل استدعاء لـ findMedian

الفكرة

خزّن كل قيمة في قائمة. ترتّبها findMedian وتقرأ الفهرسين (total-1)/2 وtotal/2، باستخدام القسمة الصحيحة: الفهرس نفسه إذا كان العدد فرديًا، والزوج الأوسط إذا كان زوجيًا. اجمع القيمتين واقسم المجموع على اثنين، فتحصل على الوسيط في الحالتين.

اقسم باستخدام الأعداد الصحيحة. يُقسم المجموع الزوجي دون باقٍ. أما المجموع الفردي فيصبح نصف قيمته المطلقة متبوعًا بـ .5، مع إشارة المجموع؛ لذا يصبح -3 هو -1.5.

تُرتّب القائمة من جديد عند كل استعلام. مع 1,500 عملية إضافة واستعلام بعد كل منها، يكون العدد نحو 10^7 مقارنة، بينما تحتاج الأكوام إلى بضع عشرات الآلاف من الخطوات.

الخوارزمية

احتفظ بقائمة values.
عند addNum x، ألحِق x وسجّل "null".
عند findMedian، رتّب values وأضف values[(total-1)/2] وvalues[total/2].
اقسم المجموع على اثنين كنص باستخدام أعداد صحيحة، وسجّله.

def half_text(total):
    # total / 2 as text, from integers only: 5 -> "2.5", -1 -> "-0.5", 6 -> "3"
    if total % 2 == 0:
        return str(total // 2)
    sign = "-" if total < 0 else ""
    return sign + str(abs(total) // 2) + ".5"


class MedianFinder:
    def __init__(self):
        self.values = []

    def addNum(self, x):
        self.values.append(x)

    def findMedian(self):
        self.values.sort()
        total = len(self.values)
        # For an odd count both indexes point at the one middle value.
        return half_text(self.values[(total - 1) // 2] + self.values[total // 2])


def medianStreamOps(ops, values):
    finder = MedianFinder()
    result = []
    for op, x in zip(ops, values):
        if op == "addNum":
            finder.addNum(x)
            result.append("null")
        else:
            result.append(finder.findMedian())
    return result

مكدسان مقسومان عند الوسيط

الفكرة

قسّم القيم إلى نصف سفلي low ونصف علوي high، بحيث تكون كل قيمة في low أقل من أو تساوي كل قيمة في high، ويكون حجم low مساويًا لحجم high أو أكبر منه بواحد. الوسيط هو أكبر قيمة في low، أو متوسطها مع أصغر قيمة في high إذا كان العدد زوجيًا. تحافظ كومة عظمى لـ low وكومة صغرى لـ high على وجود هذه القيم في القمة.

تُضاف القيمة الجديدة إلى low إذا كانت أقل من أو تساوي قيمة القمة في low، وإلا فتُضاف إلى high. عندما يختل التوازن بين حجميهما، انقل قيمة من القمة إلى النصف الآخر: فهي القيمة الأقرب إلى النصف الآخر، وبذلك يظل الترتيب صحيحًا. تتطلب عملية الإضافة بضع عمليات على الكومة، بتعقيد O(log n)، أو O(n) في R، إذ تنسخ الكومة عند كل تغيير.

بعد عمليات الإضافة الأربع في المثال الأول، تحتوي low على 3 و1، وتحتوي high على 5 و8.

الخوارزمية

احتفظ بكومة عظمى low للنصف الأدنى وكومة صغرى high للنصف الأعلى.
addNum x: أضف x إلى low إذا كانت low فارغة أو كان x أصغر من أو يساوي قمتها، وإلا فأضفه إلى high.
إذا احتوت low على قيمتين أكثر من high، فانقل قمة low إلى high. وإذا احتوت high على قيم أكثر من low، فانقل قمة high إلى low.
findMedian: إذا كانت low أكبر، فسجّل قمتها. وإلا فاجمع القمتين واقسم المجموع على اثنين باستخدام القسمة الصحيحة.

import heapq


def half_text(total):
    # total / 2 as text, from integers only: 5 -> "2.5", -1 -> "-0.5", 6 -> "3"
    if total % 2 == 0:
        return str(total // 2)
    sign = "-" if total < 0 else ""
    return sign + str(abs(total) // 2) + ".5"


class MedianFinder:
    def __init__(self):
        self.low = []   # the smaller half; heapq is a min-heap, so it stores -x to keep the largest on top
        self.high = []  # the larger half, smallest on top

    def addNum(self, x):
        if not self.low or x <= -self.low[0]:
            heapq.heappush(self.low, -x)
        else:
            heapq.heappush(self.high, x)
        # Rebalance: low holds as many values as high, or one more.
        if len(self.low) > len(self.high) + 1:
            heapq.heappush(self.high, -heapq.heappop(self.low))
        elif len(self.high) > len(self.low):
            heapq.heappush(self.low, -heapq.heappop(self.high))

    def findMedian(self):
        if len(self.low) > len(self.high):
            return str(-self.low[0])
        return half_text(-self.low[0] + self.high[0])


def medianStreamOps(ops, values):
    finder = MedianFinder()
    result = []
    for op, x in zip(ops, values):
        if op == "addNum":
            finder.addNum(x)
            result.append("null")
        else:
            result.append(finder.findMedian())
    return result

أخطاء شائعة وحالات حدّية

اختيار الكومة بحسب الحجم بدلًا من القيمة. يؤدي الإضافة إلى الكومة الأصغر من دون مقارنة إلى كسر الترتيب بين النصفين، وعندها لا تعود القيمتان في القمتين هما القيمتان الوسطيتان.
استخدام القسمة الصحيحة العادية لأخذ النصف. في C وJava، تكون نتيجة -1 / 2 هي 0، لذا فإن إلحاق .5 يطبع 0.5 للقيمة -0.5. في Python، تكون نتيجة -3 // 2 هي -2، ما يطبع -2.5 للقيمة -1.5. وتطبع الفاصلة العائمة 3.0 أو 2.500000 في بعض اللغات.
تخزين low معكوس الإشارة في كومة صغرى، ونسيان عكس الإشارة عند قراءة قيمته من القمة أو نقل قيمة إلى high.
كتابة -0. مجموع القيمتين الوسطيتين -1 و1 هو 0، والإجابة هي "0".

أسئلة شائعة3

ما التعقيد الزمني لـ Find Median from Data Stream؟

باستخدام كومتين، تكون addNum بتعقيد O(log n)، مع ثلاث عمليات دفع أو سحب من الكومة كحد أقصى، وتكون findMedian بتعقيد O(1)، أي عمليتا قراءة. تُخزَّن كل قيمة، لذا تكون المساحة O(n). أما الفرز عند كل استعلام فيكلف O(n log n) لكل عملية findMedian.

لماذا لا نحتفظ بالقيم في مصفوفة مرتبة؟

تُجيب المصفوفة المرتبة عن findMedian في O(1)، ويجد البحث الثنائي موضع القيمة الجديدة في O(log n). لكن إدراجها يزيح كل قيمة تأتي بعدها، بتعقيد O(n) لكل addNum.

هل يمكنك إيجاد الوسيط الدقيق لتدفّق دون تخزين كل قيمة؟

ليس بالضرورة. يمكن أن تصبح أي قيمة شوهدت حتى الآن هي الوسيط لاحقًا، اعتمادًا على ما يصل بعد ذلك، لذا تتطلب الإجابة الدقيقة معرفة جميع هذه القيم.

def medianStreamOps(ops, values):
    # اكتب الكود هنا

اختبار #1

اختبار #2

الوسائط

ops = ["addNum", "findMedian", "addNum", "findMedian", "addNum", "findMedian", "addNum", "findMedian"]
values = [5, 0, 1, 0, 8, 0, 3, 0]

الإخراج المتوقع

["null", "5", "null", "3", "null", "5", "null", "4"]