فهرست مقاله
۱. زمانِ رویداد را با ترتیبِ ورود یکی نگیریم
فرض کنید گزارشِ ولتاژ را از یک مسیرِ دریافت می‌خوانیم و برای هر بازهٔ زمانی، میانگین می‌خواهیم. رسیدنِ نمونهٔ تازه با برچسبِ زمانیِ بزرگ‌تر، ثابت نمی‌کند که دیگر نمونه‌ای از گذشته نخواهد رسید. اگر پنجره را زود ببندیم، دادهٔ بعدی ممکن است از خلاصه حذف شود؛ اگر همیشه منتظر بمانیم، نتیجه‌ای نهایی نداریم. این درس همین تصمیم را روی یک ورودیِ محدود و ساختگی بررسی می‌کند، نه روی حسگر یا شبکهٔ واقعی.
هر رکورد، شناسه، زمانِ رویداد برحسب میلی‌ثانیه و مقدارِ صحیحِ میلی‌ولت دارد. ترتیبِ تاپل، ترتیبِ ورود است؛ زمان‌ها می‌توانند عقب بروند یا برابر باشند. مبنای مشترکِ زمان فرض شده و زمانِ پردازش یا ساعتِ دستگاه اندازه‌گیری نمی‌شود. بخشِ نشانگرِ پیشرفت و دادهٔ دیررس در Apache Beam نیز این دو زمان و امکانِ ورودِ نامرتب را جدا توضیح می‌دهد.
در میانهٔ پنجرهٔ زمانی حسگر، زمان‌ها غیرکاهشی و پاسخ مربوط به هر پیشوندِ مرتب بود. در ادغامِ چند جریانِ مرتب نیز مرتب‌بودنِ هر منبع پیش‌شرط است. اینجا یک ورودیِ نامرتب داریم و تصمیمِ متفاوتی می‌گیریم: کدام پنجره هنوز تغییرپذیر است و چه رویدادی پس از نهایی‌شدنِ آن کنار می‌ماند؟ مرتب‌کردنِ تمامِ فایل پیش از محاسبه، این مسئله را حذف و نتیجهٔ سیاست را عوض می‌کند.
۲. نشانگرِ پیشرفت و پنجرهٔ بسته چه قراردادی دارند؟
پهنای پنجره را W می‌نامیم. زمانِ t به بازهٔ نیمه‌بازِ آغازشده از کفِ t/W ضرب‌در W تعلق دارد؛ ابتدا عضو و انتها خارج است. با پهنای ده میلی‌ثانیه، زمانِ ده عضوِ پنجرهٔ ده تا بیست است، نه صفر تا ده. شبکه از مبدأِ قراردادیِ صفر آغاز می‌شود و پنجره‌ها هم‌پوشان نیستند؛ این انتخاب با پنجرهٔ لغزانِ درسِ میانه تفاوت دارد.
سیاستِ آموزشیِ پیشرفت را F برابرِ بیشترین زمانِ رویدادِ دیده‌شده منهای L می‌گیریم. L تأخیرِ فرض‌شدهٔ پیشرفت است و لزوماً تأخیرِ شبکه یا تضمینِ سخت‌افزاری نیست. شرحِ کاملِ مولدهای نشانگر در Flink، عقب‌ماندن از بیشینهٔ زمان و وابستگیِ آن به فرضِ بی‌ترتیبی را توضیح می‌دهد. این منبع برای مفهوم خوانده شده؛ کدِ ما اجرای Flink یا بازسازیِ دقیقِ زمان‌بندیِ نشانگرهای آن نیست.
در قراردادِ خودمان، F یک مرزِ پیشرفتِ انحصاری است: پنجره‌ای با انتهای کوچک‌تر یا مساویِ F بسته می‌شود. هنگامِ ورودِ رکورد، ابتدا انتهای پنجره‌اش را با F قبلی می‌سنجیم؛ اگر آن پنجره قبلاً بسته شده، رکورد را با علت ثبت می‌کنیم و میانگینِ قبلی را تغییر نمی‌دهیم. سپس بیشینهٔ زمان و مرزِ تازه را به‌روز می‌کنیم و پنجره‌های واجدِ شرط را می‌بندیم. ترتیبِ همین دو مرحله بخشی از قرارداد است.
قدیمی‌تر بودنِ خودِ زمان از F به‌تنهایی موجبِ رد نیست؛ ممکن است پنجره هنوز باز باشد. این تفکیک را نباید از نامِ watermark حدس زد: قراردادِ مرزِ صحیحِ ما صریح است و همسانیِ API با Beam یا Flink ادعا نمی‌شود. پشتیبانی از ورودِ دیررس و بازبینیِ خروجی در چارچوب‌ها قابلِ تنظیم است؛ مثالِ حاضر عمداً خروجیِ بسته را باز نمی‌کند و منبعِ چندگانه یا تشخیصِ بی‌کاری ندارد.
۳. میانگینِ ولتاژ با دادهٔ پذیرفته‌شده تغییر می‌کند
پهنای ده و L برابرِ سه میلی‌ثانیه را انتخاب می‌کنیم. ورودِ نخست در زمانِ دو، هزار میلی‌ولت است؛ سپس نمونهٔ زمانِ دوازده با هزارودویست می‌رسد. مرزِ پیشرفت اکنون نُه است. نمونهٔ بعدی، زمانِ هشت و مقدارِ هشتصد دارد: از مرز قدیمی‌تر است، ولی انتهای پنجره‌اش ده و هنوز بزرگ‌تر از نُه است؛ پس پذیرفته می‌شود. تا اینجا میانگینِ پنجرهٔ نخست نُهصد میلی‌ولت خواهد بود.
رسیدنِ نمونهٔ زمانِ بیست‌وسه، مرز را به بیست می‌رساند. دو پنجرهٔ صفر تا ده و ده تا بیست، با میانگین‌های نُهصد و هزارودویست بسته می‌شوند. حالا نمونه‌های زمانِ پنج و نوزده می‌رسند؛ هر دو به پنجرهٔ بسته تعلق دارند و کنار می‌مانند. کنارگذاشتن در خروجی دیده می‌شود؛ نه مقدارِ صفر جای آن‌ها می‌گذاریم و نه نتیجهٔ پیشین را بی‌اعلان اصلاح می‌کنیم.
نمونه‌های زمانِ بیست‌وسه، بیست و بیست‌ونه در پنجرهٔ بیست تا سی، به‌ترتیبِ ورودِ ثبت‌شده، مقدارهای دوهزاروسیصد، دوهزار و دوهزارونهصد دارند. جمع هفت‌هزارودویست و میانگین دوهزاروچهارصد میلی‌ولت است. با ورودِ زمانِ سی‌وسه، مرز به سی می‌رسد و این پنجره نیز بسته می‌شود. نمونهٔ سی‌وسه در پنجرهٔ سی تا چهل باقی می‌ماند؛ نبودِ رویدادِ بعدی، به‌خودیِ خود آن را نهایی نمی‌کند.
این میانگین به هر مشاهده وزنِ برابر می‌دهد، نه به مدتِ نگهداریِ ولتاژ. پنجرهٔ بدونِ مشاهده را اصلاً گزارش نمی‌کنیم؛ حذفِ پنجرهٔ خالی با مشاهدهٔ معتبرِ صفر میلی‌ولت فرق دارد. اگر پیش از پردازش همهٔ رکوردها را مرتب کنیم، دو دادهٔ کنارگذاشته‌شده واردِ گروه‌های قبلی می‌شوند و میانگینِ نخست به ۵۸۰۰/۳ می‌رسد. این عدد، خروجیِ سیاستِ دیگری است؛ نه صرفاً نمایشِ مرتب‌ترِ همان گزارش.
۴. کدِ کامل و مرجعِ مستقلِ پیشوندها
فایلِ event_time_watermark.py فقط کتابخانهٔ استانداردِ پایتون می‌خواهد. ورودی، tuple واقعی با حداکثر سی‌ودو رکوردِ سه‌عضوی است؛ شناسهٔ ASCII کوتاه و یکتا، زمانِ int واقعی از صفر تا هزار میلی‌ثانیه و مقدارِ int واقعی در بازهٔ منفی پنج‌هزار تا پنج‌هزار میلی‌ولت. پهنا از یک تا صد و L از صفر تا صد است. این کران‌ها سیاستِ آموزشی‌اند؛ اعتبارسنجی، واحدِ فیزیکی یا درستیِ ساعتِ ورودی را کشف نمی‌کند.
"""Synthetic arrival-order voltage windows; not a Beam/Flink or live clock."""
from fractions import Fraction
from itertools import product
import json
import platform
import re
def checked_profile(events, width_ms, delay_ms, finish):
if type(events) is not tuple or len(events) > 32:
raise TypeError('Exact tuple of at most32 arrival-ordered events required')
if type(width_ms) is not int or not 1 <= width_ms <= 100:
raise ValueError('Exact integer window width1..100ms required')
if type(delay_ms) is not int or not 0 <= delay_ms <= 100:
raise ValueError('Exact integer assumed progress delay0..100ms required')
if type(finish) is not bool:
raise TypeError('Exact bool finite-input completion declaration required')
seen = set()
for event in events:
if type(event) is not tuple or len(event) != 3:
raise TypeError('Exact(id,event_time_ms,voltage_mV) tuple required')
key, at, value = event
if type(key) is not str or re.fullmatch(r'[A-Za-z0-9_-]{1,12}', key) is None:
raise ValueError('ASCII identifier of1..12 characters required')
if key in seen:
raise ValueError('Unique event identifiers required; no replay policy here')
seen.add(key)
if type(at) is not int or not 0 <= at <= 1000:
raise ValueError('Exact integer event timestamp0..1000ms required')
if type(value) is not int or not -5000 <= value <= 5000:
raise ValueError('Exact integer synthetic voltage within[-5000,5000]mV required')
def window_report(events, width_ms, delay_ms, finish=False):
"""Own exclusive progress frontier, not an inclusive framework watermark.
F=max seen event timestamp-delay; finalize half-open windows if end<=F.
Refuse observations only when their window already lies behind F. No
processing-time clock, source-completeness inference or output revision.
"""
checked_profile(events, width_ms, delay_ms, finish)
maximum = frontier = None
pending = {}
emitted, refused, accepted_ids, trace = [], [], [], []
def summary(start, state, reason, position, closing_frontier):
total, count, identities = state
return {'window_ms': [start, start + width_ms], 'count': count,
'sum_mV': total, 'mean_mV_exact': str(Fraction(total, count)),
'event_ids_in_arrival_order': list(identities), 'reason': reason,
'emitted_after_arrival_index': position,
'closing_frontier_ms': closing_frontier}
for position, (key, at, value) in enumerate(events):
start = at // width_ms * width_ms
previous = frontier
is_late = previous is not None and start + width_ms <= previous
if is_late:
refused.append({'id': key, 'event_time_ms': at, 'voltage_mV': value,
'window_ms': [start, start + width_ms],
'previous_frontier_ms': previous})
else:
state = pending.setdefault(start, [0, 0, []])
state[0] += value; state[1] += 1; state[2].append(key)
accepted_ids.append(key)
maximum = at if maximum is None else max(maximum, at)
frontier = maximum - delay_ms
closed = sorted(s for s in pending if s + width_ms <= frontier)
for s in closed:
emitted.append(summary(s, pending.pop(s), 'frontier', position, frontier))
trace.append({'arrival_index': position, 'id': key,
'previous_frontier_ms': previous, 'frontier_ms': frontier,
'refused_closed_window': is_late, 'closed_window_starts_ms': closed})
if finish:
for start in sorted(pending):
emitted.append(summary(start, pending[start], 'declared_end_of_input', len(events), None))
pending.clear()
return {'emitted': emitted,
'pending': [summary(s, pending[s], 'pending', None, None) for s in sorted(pending)],
'accepted_event_ids': accepted_ids, 'refused': refused,
'arrival_trace': trace, 'final_frontier_ms': frontier,
'source_complete_declared': finish}
def prefix_reference(events, width, delay, finish):
"""Trusted authored fixtures only. Whole prefix scans, no main accumulator.
A window closes at the first prefix maximum >= its end+delay. Membership
is decided from the preceding prefix maximum. All groups are built by
independent full scans; no pending-state mutation or main helper call.
"""
maxima = [max(t for _, t, _ in events[:i+1]) for i in range(len(events))]
fronts = [t-delay for t in maxima]
accepted, refused = [], []
for i, (key, t, value) in enumerate(events):
start = t // width * width
old_front = fronts[i-1] if i else None
if old_front is None or start + width > old_front:
accepted.append(i)
else:
refused.append({'id': key, 'event_time_ms': t, 'voltage_mV': value,
'window_ms': [start, start+width], 'previous_frontier_ms': old_front})
starts = sorted({events[i][1] // width * width for i in accepted})
emitted, pending, closure = [], [], {}
for start in starts:
indices = [i for i in accepted if start <= events[i][1] < start+width]
closes = [i for i, maximum in enumerate(maxima) if maximum >= start+width+delay]
at = min(closes) if closes else None
closure[start] = at
total = sum(events[i][2] for i in indices)
row = {'window_ms': [start, start+width], 'count': len(indices),
'sum_mV': total, 'mean_mV_exact': str(Fraction(total, len(indices))),
'event_ids_in_arrival_order': [events[i][0] for i in indices],
'reason': 'frontier' if at is not None else ('declared_end_of_input' if finish else 'pending'),
'emitted_after_arrival_index': at if at is not None else (len(events) if finish else None),
'closing_frontier_ms': fronts[at] if at is not None else None}
(emitted if at is not None or finish else pending).append(row)
emitted.sort(key=lambda row: (row['emitted_after_arrival_index'], row['window_ms'][0]))
refused_ids = {row['id'] for row in refused}
trace = [{'arrival_index': i, 'id': key, 'previous_frontier_ms': fronts[i-1] if i else None,
'frontier_ms': fronts[i], 'refused_closed_window': key in refused_ids,
'closed_window_starts_ms': [s for s in starts if closure[s] == i]}
for i, (key, _, _) in enumerate(events)]
return {'emitted': emitted, 'pending': pending,
'accepted_event_ids': [events[i][0] for i in accepted], 'refused': refused,
'arrival_trace': trace, 'final_frontier_ms': fronts[-1] if fronts else None,
'source_complete_declared': finish}
def refuses(*args):
try:
window_report(*args)
except (TypeError, ValueError):
return True
return False
def run():
events = (('A',2,1000),('B',12,1200),('C',8,800),('D',23,2300),
('E',5,4000),('F',19,1400),('G',20,2000),('H',29,2900),('I',33,3300))
open_result = window_report(events, 10, 3)
completed = window_report(events, 10, 3, True)
checks = {}
def check(name, value):
assert type(value) is bool
checks[name] = value
assert value, name
check('hand_three_finalized_windows_two_refusals_and_one_pending_mean',
[(x['window_ms'],x['mean_mV_exact'],x['emitted_after_arrival_index']) for x in open_result['emitted']]
== [([0,10],'900',3),([10,20],'1200',3),([20,30],'2400',8)]
and [x['id'] for x in open_result['refused']] == ['E','F']
and [(x['window_ms'],x['mean_mV_exact']) for x in open_result['pending']] == [([30,40],'3300')])
check('older_than_frontier_can_enter_a_still_open_window',
open_result['arrival_trace'][2]['previous_frontier_ms'] == 9
and events[2][1] == 8 and 'C' in open_result['accepted_event_ids']
and open_result['arrival_trace'][2]['refused_closed_window'] is False)
check('end_equal_to_frontier_closes_and_boundary_timestamp_belongs_to_next_window',
open_result['arrival_trace'][3]['frontier_ms'] == 20
and open_result['arrival_trace'][3]['closed_window_starts_ms'] == [0,10]
and 'G' in open_result['emitted'][2]['event_ids_in_arrival_order']
and open_result['emitted'][2]['window_ms'] == [20,30])
check('declared_input_end_flushes_pending_without_inventing_a_numeric_frontier',
completed['emitted'][:3] == open_result['emitted'] and completed['pending'] == []
and completed['emitted'][-1]['reason'] == 'declared_end_of_input'
and completed['emitted'][-1]['closing_frontier_ms'] is None
and completed['final_frontier_ms'] == open_result['final_frontier_ms'] == 30)
sorted_result = window_report(tuple(sorted(events,key=lambda e:e[1])),10,3,True)
check('sorting_all_arrivals_first_changes_the_policy_result_not_just_presentation',
sorted_result['refused'] == [] and sorted_result['emitted'][0]['mean_mV_exact'] == '5800/3'
and sorted_result['emitted'][1]['mean_mV_exact'] == '1300'
and completed['emitted'][0]['mean_mV_exact'] == '900')
wider = window_report(events,10,100,True)
check('larger_assumed_delay_retains_more_observations_not_a_physical_accuracy_guarantee',
wider['refused'] == [] and wider['final_frontier_ms'] == -67
and set(completed['accepted_event_ids']) <= set(wider['accepted_event_ids'])
and all(x['reason']=='declared_end_of_input' for x in wider['emitted']))
check('finite_prefix_outputs_do_not_use_future_data_or_revise_closed_results',
all(window_report(events[:n],10,3)['emitted'] == [x for x in open_result['emitted'] if x['emitted_after_arrival_index'] < n]
for n in range(len(events)+1))
and window_report(events+(('J',0,-5000),),10,3)['emitted'] == open_result['emitted'])
same = window_report((('a',10,1000),('b',10,3000)),10,0,True)
gap = window_report((('a',0,0),('b',40,0)),10,0,True)
check('same_time_observations_are_distinct_and_empty_windows_are_not_fabricated_zero',
len(same['emitted'])==1 and same['emitted'][0]['count']==2 and same['emitted'][0]['mean_mV_exact']=='2000'
and [x['window_ms'] for x in gap['emitted']]==[[0,10],[40,50]]
and all(x['mean_mV_exact']=='0' for x in gap['emitted']))
values = {key:value for key,_,value in events}
check('accepted_refused_partition_and_integer_sum_count_conservation',
sum(x['count'] for x in completed['emitted']) + len(completed['refused']) == len(events)
and sum(x['sum_mV'] for x in completed['emitted']) == sum(values[k] for k in completed['accepted_event_ids']) == 13500)
stretched = window_report(tuple((k,2*t,v) for k,t,v in events),20,6,True)
check('physical_time_dilation_in_unchanged_ms_preserves_membership_and_voltage_means',
[(x['window_ms'],x['mean_mV_exact'],x['event_ids_in_arrival_order']) for x in stretched['emitted']]
== [([2*t for t in x['window_ms']],x['mean_mV_exact'],x['event_ids_in_arrival_order']) for x in completed['emitted']]
and stretched['final_frontier_ms']==2*completed['final_frontier_ms'])
inverted = window_report(tuple((k,t,-v) for k,t,v in events),10,3,True)
shifted = window_report(tuple((k,t,v+100) for k,t,v in events),10,3,True)
check('voltage_sign_and_common_offset_change_means_not_acceptance',
inverted['accepted_event_ids']==shifted['accepted_event_ids']==completed['accepted_event_ids']
and [Fraction(x['mean_mV_exact']) for x in inverted['emitted']]==[-Fraction(x['mean_mV_exact']) for x in completed['emitted']]
and [Fraction(x['mean_mV_exact']) for x in shifted['emitted']]==[Fraction(x['mean_mV_exact'])+100 for x in completed['emitted']])
cases=0
for n in range(5):
for times in product(range(5),repeat=n):
rows=tuple((f'E{i}',t,(-1)**i*(i+1)) for i,t in enumerate(times))
for width,delay,finish in product((1,2,3),(0,1,2),(False,True)):
assert window_report(rows,width,delay,finish)==prefix_reference(rows,width,delay,finish)
cases+=1
check('independent_prefix_maximum_first_closure_fullscan_oracle_ONE_family', cases==14058)
empty=window_report((),1,0,True)
cap=window_report(tuple((f'E{i}',1000,(-5000 if i%2 else 5000)) for i in range(32)),100,100,True)
check('empty_zero_delay_and_inclusive_domain_endpoints_are_explicit',
empty['emitted']==empty['pending']==empty['refused']==[] and empty['final_frontier_ms'] is None
and cap['emitted'][0]['window_ms']==[1000,1100] and cap['emitted'][0]['count']==32
and cap['emitted'][0]['mean_mV_exact']=='0')
class IntChild(int):pass
class TupleChild(tuple):
def __len__(self):raise AssertionError('Excluded hooks must not execute')
bad=[(list(events),10,3,False),(TupleChild(events),10,3,False),((('a',True,1),),10,3,False),
((('a',1,IntChild(1)),),10,3,False),((('a',1,1.0),),10,3,False),((('a',-1,0),),10,3,False),
((('a',1001,0),),10,3,False),((('a',0,5001),),10,3,False),((('a',0,0),('a',1,1)),10,3,False),
((('bad id',0,0),),10,3,False),((('a',0,0),),True,3,False),(events,0,3,False),(events,101,3,False),
(events,10,-1,False),(events,10,101,False),(events,10,False,False),(events,10,3,0),
(tuple((f'E{i}',i,0) for i in range(33)),10,3,False),((['a',0,0],),10,3,False)]
check('whole_profile_exact_types_IDs_shape_and_bounds_rejected_before_accumulation', all(refuses(*args) for args in bad))
check('repeat_calls_and_original_arrival_order_inputs_remain_unchanged',
window_report(events,10,3)==open_result and events[2]==('C',8,800) and len(events)==9)
return {'python':platform.python_version(),'implementation':platform.python_implementation(),
'dependencies':'standard library only; Beam and Flink not executed',
'units':{'event_time':'relative integer ms on an assumed common clock','voltage':'synthetic integer mV','mean':'exact observation-weighted mV, not duration weighted'},
'policy':{'exclusive_progress_frontier':'F=max observed event-time-delay_ms','half_open_windows':'[floor(t/W)*W, floor(t/W)*W+W)','finalize':'nonempty window end<=F','refuse':'incoming window already end<=previousF','EOF':'explicit source-complete declaration, not elapsed wall time','framework_API_parity_claimed':False},
'width_ms':10,'assumed_delay_ms':3,'arrival_order':events,
'without_end_declaration':open_result,'with_explicit_end_declaration':completed,
'sort_before_processing_first_mean_mV':sorted_result['emitted'][0]['mean_mV_exact'],
'large_delay_all_observations_first_mean_mV':wider['emitted'][0]['mean_mV_exact'],
'reference_scenarios_inside_ONE_family':cases,'checks':checks,'unique_check_groups':len(checks),
'limitations':['One finite synthetic input, no IO, timers, multiple input watermarks, idle-source or persistent exactly-once state.',
'The lateness/frontier policy does not establish a real delay bound, common-clock accuracy or complete physical acquisition.',
'Dropped data are disclosed; emitted means are not revised. Empty windows are omitted, not observed zero voltage.',
'Exact integer/Fraction arithmetic does not remove measurement uncertainty or select a production watermark policy.']}
if __name__=='__main__':
print(json.dumps(run(),ensure_ascii=True,indent=2,allow_nan=False))
مرجعِ مستقل برای هر پنجره تمامِ پیشوندها را می‌گردد: نخستین جایی که بیشینهٔ زمان به انتهای پنجره به‌اضافهٔ L برسد، جای بسته‌شدن است. پذیرفته‌شدنِ هر رکورد نیز از بیشینهٔ پیشوندِ قبلی تعیین می‌شود. مرجع، انباشتگر یا تابعِ اصلی را صدا نمی‌زند. هزاران سناریوی محدودِ ترتیبِ ورود، پهنا، تأخیر و پایانِ ورودی، یک خانوادهٔ بررسی‌اند؛ تعدادِ سناریوها تعدادِ اجرای مستقل یا گروه‌های نام‌دار نیست.
جمع و شمارش صحیح‌اند و میانگین با Fraction نوشته می‌شود؛ بنابراین گردکردنِ ممیزشناور واردِ این مدل نمی‌شود. این ویژگی، عدمِ قطعیتِ اندازه‌گیری یا دادهٔ حذف‌شده را جبران نمی‌کند. بررسی‌های دیگر، مرزِ مساوی، پنجرهٔ خالی، زمانِ برابر، حفظِ پیشوند، ترازِ جمع و شمار، تغییرِ مقدار و کشیدگیِ فیزیکیِ زمان با حفظِ واحد را جدا می‌سنجند. کلِ ورودی پیش از انباشت اعتبارسنجی می‌شود و حالتِ بیرونی تغییر نمی‌کند.
{
"python": "3.14.5",
"implementation": "CPython",
"dependencies": "standard library only; Beam and Flink not executed",
"units": {
"event_time": "relative integer ms on an assumed common clock",
"voltage": "synthetic integer mV",
"mean": "exact observation-weighted mV, not duration weighted"
},
"policy": {
"exclusive_progress_frontier": "F=max observed event-time-delay_ms",
"half_open_windows": "[floor(t/W)*W, floor(t/W)*W+W)",
"finalize": "nonempty window end<=F",
"refuse": "incoming window already end<=previousF",
"EOF": "explicit source-complete declaration, not elapsed wall time",
"framework_API_parity_claimed": false
},
"width_ms": 10,
"assumed_delay_ms": 3,
"arrival_order": [
[
"A",
2,
1000
],
[
"B",
12,
1200
],
[
"C",
8,
800
],
[
"D",
23,
2300
],
[
"E",
5,
4000
],
[
"F",
19,
1400
],
[
"G",
20,
2000
],
[
"H",
29,
2900
],
[
"I",
33,
3300
]
],
"without_end_declaration": {
"emitted": [
{
"window_ms": [
0,
10
],
"count": 2,
"sum_mV": 1800,
"mean_mV_exact": "900",
"event_ids_in_arrival_order": [
"A",
"C"
],
"reason": "frontier",
"emitted_after_arrival_index": 3,
"closing_frontier_ms": 20
},
{
"window_ms": [
10,
20
],
"count": 1,
"sum_mV": 1200,
"mean_mV_exact": "1200",
"event_ids_in_arrival_order": [
"B"
],
"reason": "frontier",
"emitted_after_arrival_index": 3,
"closing_frontier_ms": 20
},
{
"window_ms": [
20,
30
],
"count": 3,
"sum_mV": 7200,
"mean_mV_exact": "2400",
"event_ids_in_arrival_order": [
"D",
"G",
"H"
],
"reason": "frontier",
"emitted_after_arrival_index": 8,
"closing_frontier_ms": 30
}
],
"pending": [
{
"window_ms": [
30,
40
],
"count": 1,
"sum_mV": 3300,
"mean_mV_exact": "3300",
"event_ids_in_arrival_order": [
"I"
],
"reason": "pending",
"emitted_after_arrival_index": null,
"closing_frontier_ms": null
}
],
"accepted_event_ids": [
"A",
"B",
"C",
"D",
"G",
"H",
"I"
],
"refused": [
{
"id": "E",
"event_time_ms": 5,
"voltage_mV": 4000,
"window_ms": [
0,
10
],
"previous_frontier_ms": 20
},
{
"id": "F",
"event_time_ms": 19,
"voltage_mV": 1400,
"window_ms": [
10,
20
],
"previous_frontier_ms": 20
}
],
"arrival_trace": [
{
"arrival_index": 0,
"id": "A",
"previous_frontier_ms": null,
"frontier_ms": -1,
"refused_closed_window": false,
"closed_window_starts_ms": []
},
{
"arrival_index": 1,
"id": "B",
"previous_frontier_ms": -1,
"frontier_ms": 9,
"refused_closed_window": false,
"closed_window_starts_ms": []
},
{
"arrival_index": 2,
"id": "C",
"previous_frontier_ms": 9,
"frontier_ms": 9,
"refused_closed_window": false,
"closed_window_starts_ms": []
},
{
"arrival_index": 3,
"id": "D",
"previous_frontier_ms": 9,
"frontier_ms": 20,
"refused_closed_window": false,
"closed_window_starts_ms": [
0,
10
]
},
{
"arrival_index": 4,
"id": "E",
"previous_frontier_ms": 20,
"frontier_ms": 20,
"refused_closed_window": true,
"closed_window_starts_ms": []
},
{
"arrival_index": 5,
"id": "F",
"previous_frontier_ms": 20,
"frontier_ms": 20,
"refused_closed_window": true,
"closed_window_starts_ms": []
},
{
"arrival_index": 6,
"id": "G",
"previous_frontier_ms": 20,
"frontier_ms": 20,
"refused_closed_window": false,
"closed_window_starts_ms": []
},
{
"arrival_index": 7,
"id": "H",
"previous_frontier_ms": 20,
"frontier_ms": 26,
"refused_closed_window": false,
"closed_window_starts_ms": []
},
{
"arrival_index": 8,
"id": "I",
"previous_frontier_ms": 26,
"frontier_ms": 30,
"refused_closed_window": false,
"closed_window_starts_ms": [
20
]
}
],
"final_frontier_ms": 30,
"source_complete_declared": false
},
"with_explicit_end_declaration": {
"emitted": [
{
"window_ms": [
0,
10
],
"count": 2,
"sum_mV": 1800,
"mean_mV_exact": "900",
"event_ids_in_arrival_order": [
"A",
"C"
],
"reason": "frontier",
"emitted_after_arrival_index": 3,
"closing_frontier_ms": 20
},
{
"window_ms": [
10,
20
],
"count": 1,
"sum_mV": 1200,
"mean_mV_exact": "1200",
"event_ids_in_arrival_order": [
"B"
],
"reason": "frontier",
"emitted_after_arrival_index": 3,
"closing_frontier_ms": 20
},
{
"window_ms": [
20,
30
],
"count": 3,
"sum_mV": 7200,
"mean_mV_exact": "2400",
"event_ids_in_arrival_order": [
"D",
"G",
"H"
],
"reason": "frontier",
"emitted_after_arrival_index": 8,
"closing_frontier_ms": 30
},
{
"window_ms": [
30,
40
],
"count": 1,
"sum_mV": 3300,
"mean_mV_exact": "3300",
"event_ids_in_arrival_order": [
"I"
],
"reason": "declared_end_of_input",
"emitted_after_arrival_index": 9,
"closing_frontier_ms": null
}
],
"pending": [],
"accepted_event_ids": [
"A",
"B",
"C",
"D",
"G",
"H",
"I"
],
"refused": [
{
"id": "E",
"event_time_ms": 5,
"voltage_mV": 4000,
"window_ms": [
0,
10
],
"previous_frontier_ms": 20
},
{
"id": "F",
"event_time_ms": 19,
"voltage_mV": 1400,
"window_ms": [
10,
20
],
"previous_frontier_ms": 20
}
],
"arrival_trace": [
{
"arrival_index": 0,
"id": "A",
"previous_frontier_ms": null,
"frontier_ms": -1,
"refused_closed_window": false,
"closed_window_starts_ms": []
},
{
"arrival_index": 1,
"id": "B",
"previous_frontier_ms": -1,
"frontier_ms": 9,
"refused_closed_window": false,
"closed_window_starts_ms": []
},
{
"arrival_index": 2,
"id": "C",
"previous_frontier_ms": 9,
"frontier_ms": 9,
"refused_closed_window": false,
"closed_window_starts_ms": []
},
{
"arrival_index": 3,
"id": "D",
"previous_frontier_ms": 9,
"frontier_ms": 20,
"refused_closed_window": false,
"closed_window_starts_ms": [
0,
10
]
},
{
"arrival_index": 4,
"id": "E",
"previous_frontier_ms": 20,
"frontier_ms": 20,
"refused_closed_window": true,
"closed_window_starts_ms": []
},
{
"arrival_index": 5,
"id": "F",
"previous_frontier_ms": 20,
"frontier_ms": 20,
"refused_closed_window": true,
"closed_window_starts_ms": []
},
{
"arrival_index": 6,
"id": "G",
"previous_frontier_ms": 20,
"frontier_ms": 20,
"refused_closed_window": false,
"closed_window_starts_ms": []
},
{
"arrival_index": 7,
"id": "H",
"previous_frontier_ms": 20,
"frontier_ms": 26,
"refused_closed_window": false,
"closed_window_starts_ms": []
},
{
"arrival_index": 8,
"id": "I",
"previous_frontier_ms": 26,
"frontier_ms": 30,
"refused_closed_window": false,
"closed_window_starts_ms": [
20
]
}
],
"final_frontier_ms": 30,
"source_complete_declared": true
},
"sort_before_processing_first_mean_mV": "5800/3",
"large_delay_all_observations_first_mean_mV": "5800/3",
"reference_scenarios_inside_ONE_family": 14058,
"checks": {
"hand_three_finalized_windows_two_refusals_and_one_pending_mean": true,
"older_than_frontier_can_enter_a_still_open_window": true,
"end_equal_to_frontier_closes_and_boundary_timestamp_belongs_to_next_window": true,
"declared_input_end_flushes_pending_without_inventing_a_numeric_frontier": true,
"sorting_all_arrivals_first_changes_the_policy_result_not_just_presentation": true,
"larger_assumed_delay_retains_more_observations_not_a_physical_accuracy_guarantee": true,
"finite_prefix_outputs_do_not_use_future_data_or_revise_closed_results": true,
"same_time_observations_are_distinct_and_empty_windows_are_not_fabricated_zero": true,
"accepted_refused_partition_and_integer_sum_count_conservation": true,
"physical_time_dilation_in_unchanged_ms_preserves_membership_and_voltage_means": true,
"voltage_sign_and_common_offset_change_means_not_acceptance": true,
"independent_prefix_maximum_first_closure_fullscan_oracle_ONE_family": true,
"empty_zero_delay_and_inclusive_domain_endpoints_are_explicit": true,
"whole_profile_exact_types_IDs_shape_and_bounds_rejected_before_accumulation": true,
"repeat_calls_and_original_arrival_order_inputs_remain_unchanged": true
},
"unique_check_groups": 15,
"limitations": [
"One finite synthetic input, no IO, timers, multiple input watermarks, idle-source or persistent exactly-once state.",
"The lateness/frontier policy does not establish a real delay bound, common-clock accuracy or complete physical acquisition.",
"Dropped data are disclosed; emitted means are not revised. Empty windows are omitted, not observed zero voltage.",
"Exact integer/Fraction arithmetic does not remove measurement uncertainty or select a production watermark policy."
]
}نسخهٔ واقعیِ اجرا همراهِ خروجی ثبت است. اسنادِ خوانده‌شده Flink 2.3.0 را نشان می‌دهند؛ نسخهٔ SDK برای صفحهٔ عمومیِ Beam تعیین نشده و هیچ‌کدام اجرا نشده‌اند. تاریخِ دسترسی ۲۸ مهر ۱۴۰۵ است؛ زمانِ تغییرِ فایلِ HTTP را تاریخِ نخستین انتشار نمی‌گیریم. این تمرین آفلاین است: از سکوتِ ورودی، موعدِ واقعی، دوامِ ذخیره‌سازی یا تضمینِ پردازشِ دقیقاً یک‌بار نتیجه نمی‌گیریم.
۵. چهار پرسش دربارهٔ دیررسیدن و نهایی‌شدن
پرسش اول: چرا نمونهٔ زمانِ هشت با مرزِ نُه پذیرفته شد؟ پاسخ: چون معیارِ رد، بسته‌بودنِ پنجره است؛ انتهای پنجرهٔ آن ده و هنوز از نُه بزرگ‌تر بود. ردکردنِ هر زمانِ کوچک‌تر از مرز، قراردادِ محافظه‌کارانهٔ دیگری می‌سازد و همین داده را حذف می‌کند؛ نامِ پارامتر جایِ شرطِ دقیق را نمی‌گیرد.
پرسش دوم: نمونهٔ دقیقاً در زمانِ بیست به کجا می‌رود؟ پاسخ: به پنجرهٔ بیست تا سی. بسته‌شدنِ پنجرهٔ ده تا بیست در مرزِ بیست مانعِ پذیرشِ آن نیست. بازکردنِ انتهای پنجره و بستنِ ابتدای بعدی باعث می‌شود یک مشاهده در دو پنجره شمرده نشود؛ کد این تساوی را مستقل بررسی می‌کند.
پرسش سوم: بزرگ‌تر کردنِ L همیشه نتیجه را دقیق‌تر می‌کند؟ پاسخ: در این سیاست می‌تواند دادهٔ بیشتری را نگه دارد و نهایی‌شدن را عقب بیندازد، اما دقتِ فیزیکی یا صحتِ رویدادها را ثابت نمی‌کند. میانگین با ورودِ مقادیرِ تازه ممکن است در هر جهت تغییر کند. انتخابِ L به قراردادِ منبع و نیازِ کاربرد وابسته است؛ سه میلی‌ثانیه توصیهٔ عملیاتی نیست.
پرسش چهارم: گزینهٔ finish یعنی مدتی صبر کرده‌ایم؟ پاسخ: خیر؛ اعلامِ صریحِ کامل‌بودنِ همین ورودیِ محدود است. پنجرهٔ آخر با علتِ جداگانهٔ پایانِ ورودی تخلیه می‌شود و مرزِ عددیِ ساختگی به بی‌نهایت نمی‌رود. پایانِ فهرستِ آزمایش، پایانِ حسگرِ زنده را ثابت نمی‌کند؛ بدونِ این اعلام، همان پنجره در خروجیِ معلق می‌ماند.
منابع و مطالعهٔ بیشتر
- Apache Flink2.3.0؛ مولدهای نشانگرِ پیشرفت و عقب‌ماندگیِ ثابت
- Apache Beam؛ بخشِ کاملِ نشانگرِ پیشرفت و دادهٔ دیررس
بررسی منابع: ۲۸ مهر ۱۴۰۵. تحلیل‌ها و پیشنهادهای اجرایی، جمع‌بندی نویسنده‌اند.
