Find Median from Data Stream
2つの操作を持つ構造を設計してください。addNumは整数を保存し、findMedianはそれまでに保存されたすべての値の中央値を返します。個数が奇数の場合、中央値は中央の値です。偶数の場合は、中央にある2つの値の平均です。
操作iはops[i]です。values[i]には、addNumの場合は数値が、findMedianの場合は0が入っています。空の構造に対して順番に操作を実行し、それぞれについて文字列を1つ返してください。addNumの場合は"null"、findMedianの場合は中央値を返します。中央値は"3"、"2.5"、"-0.5"のように表記してください。整数の場合は小数点を付けず、ゼロは"0"とし、"-0"にはしないでください。
関数
- opsstring-array
- 実行される順序での操作
- valuesinteger-array
- 各addNumに対する数値、各findMedianに対しては0
- 戻り値string-array
- 操作ごとに1つの回答をテキスト形式で
制約
1 ≤ ops.length ≤ 3000values.length == ops.length- 各
ops[i]はaddNumまたはfindMedianです。 -105 ≤ values[i] ≤ 105はaddNumの場合、values[i] == 0はfindMedianの場合です。findMedianは、少なくとも 1 回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 です。
- 入力
- ops = ["addNum", "addNum", "findMedian", "addNum", "addNum", "findMedian"]values = [-4, 3, 0, -2, -1, 0]
- 出力
- ["null", "null", "-0.5", "null", "null", "-1.5"]
- 説明
- -4 と 3 の合計は -1 なので、中央値は -0.5 です。-2 と -1 の後では、並べ替えられた値は -4、-2、-1、3 となります。中央の 2 つの値の合計は -3 なので、中央値は -1.5 です。
発展問題
すべての値が0から100の範囲内だとしたら、両方の操作を定数時間で実行するにはどうすればよいでしょうか？
ヒント
1つずつ開いてください。開くたびに少しずつ答えに近づきます。
findMedianを呼び出すたびにすべての値をソートする方法は正しいですが、多くの作業を繰り返します。中央値は真ん中だけを見ます。すばやく見つける必要がある値はどれでしょうか？値を下半分と上半分に分けます。中央値は、下半分の最大値と上半分の最小値だけで決まり、それぞれの半分の中での順序は関係ありません。
下半分を最大ヒープ、上半分を最小ヒープに保持し、下半分の値の数を上半分と同じか、1つ多くします。新しい値は下半分の先頭と比較して適切な側に入れ、サイズのバランスが崩れたら、先頭の値を1つ反対側に移します。
解説
中央値はデータの2つの半分の間に位置するため、それぞれの半分の中での順序は重要ではありません。重要なのは、下半分の最大値と上半分の最小値だけです。2つのヒープを使うと、この2つの値がそれぞれヒープの先頭に来ます。各 addNum のコストは O(log n) で、各 findMedian は O(1) で先頭の2つの値を読み取ります。
findMedian のたびにソートする
考え方
すべての値をリストに格納します。findMedianで並べ替え、整数除算でインデックス(total-1)/2とtotal/2を読み取ります。個数が奇数なら同じインデックス、偶数なら中央の2つの値になります。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中央値で分割された2つのヒープ
考え方
値を下半分のlowと上半分のhighに分けます。lowのすべての値はhighのすべての値以下とし、lowのサイズはhighと同じか、1つ大きくします。中央値はlowの最大値です。個数が偶数の場合は、lowの最大値とhighの最小値の平均です。lowには最大ヒープ、highには最小ヒープを使うことで、それぞれの値を先頭に保ちます。
新しい値がlowの先頭以下ならlowに、そうでなければhighに入れます。サイズの差が大きくなったら、先頭の値をもう一方へ移します。その値はもう一方の半分に最も近いため、順序が保たれます。追加処理はヒープ操作を数回行うだけなのでO(log n)ですが、Rでは変更のたびにヒープがコピーされるためO(n)です。
最初の例で4回追加した後、lowには3と1が、highには5と8が入っています。
アルゴリズム
- 下半分用の最大ヒープ
lowと、上半分用の最小ヒープhighを保持します。 addNum x:lowが空、またはxがその先頭以下なら、xをlowに追加します。それ以外の場合はhighに追加します。lowの値がhighより2つ多い場合は、lowの先頭をhighに移します。highの値がlowより多い場合は、highの先頭をlowに移します。findMedian:lowのほうが大きければ、その先頭を記録します。それ以外の場合は、両方の先頭を足し、その合計を整数で2で割ります。
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
落とし穴と境界ケース
- 値ではなくサイズでヒープを選ぶ。比較せずに小さい方のヒープに追加すると、2つの半分の間の順序が崩れ、ヒープの先頭が中央の値ではなくなる。
- 通常の整数除算で半分にする。C と Java では
-1 / 2は 0 なので、.5を付け加えると -0.5 に対して 0.5 が出力される。Python では-3 // 2は -2 なので、-1.5 に対して -2.5 が出力される。言語によっては浮動小数点数で 3.0 や 2.500000 と出力される。 lowを最小ヒープ内で符号を反転して格納し、先頭を読み取るときや値をhighに移すときに、符号を元に戻し忘れる。-0と書く。中央の値 -1 と 1 の合計は 0 なので、答えは"0"。
よくある質問3
データストリームから中央値を求める場合の時間計算量はどれくらいですか？
2つのヒープを使うと、addNumはO(log n)で、ヒープへの追加または取り出しは最大3回です。また、findMedianはO(1)で、読み取りは2回です。すべての値を保存するため、空間計算量はO(n)です。代わりに、クエリのたびにソートすると、findMedianごとにO(n log n)のコストがかかります。
値をソート済み配列に保持しておけばいいのではないでしょうか？
ソート済み配列ならfindMedianはO(1)で求められ、二分探索を使えば新しい値を挿入する位置をO(log n)で見つけられます。しかし、挿入するとその位置以降のすべての値がずれるため、addNumごとにO(n)かかります。
すべての値を保存せずに、ストリームの正確な中央値を求められますか？
一般的にはそうではありません。これまでに見たどの値も、次に何が到着するかによっては後で中央値になる可能性があるため、正確な答えを出すにはそれらすべてが必要です。
似た問題
同じ考え方を使う問題です。2〜3問解くとパターンが身につきます。
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"]