-
Notifications
You must be signed in to change notification settings - Fork 4.2k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
[BEAM-9488] Ensure we pass through PCollection ids instead of attempting to fix them up. #11514
Changes from 1 commit
96cc611
808b4b9
04c6334
051c094
4c03658
1297a0b
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
…ing to fix them up.
- Loading branch information
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -1035,59 +1035,12 @@ def monitoring_infos(self): | |
# Construct a new dict first to remove duplicates. | ||
all_monitoring_infos_dict = {} | ||
for transform_id, op in self.ops.items(): | ||
for mi in op.monitoring_infos(transform_id).values(): | ||
fixed_mi = self._fix_output_tags_monitoring_info(transform_id, mi) | ||
all_monitoring_infos_dict[monitoring_infos.to_key(fixed_mi)] = fixed_mi | ||
|
||
infos_list = list(all_monitoring_infos_dict.values()) | ||
|
||
def inject_pcollection(monitoring_info): | ||
""" | ||
If provided metric is element count metric: | ||
Finds relevant transform output info in current process_bundle_descriptor | ||
and adds tag with PCOLLECTION_LABEL and pcollection_id into monitoring | ||
info. | ||
""" | ||
if monitoring_info.urn in URNS_NEEDING_PCOLLECTIONS: | ||
if not monitoring_infos.PTRANSFORM_LABEL in monitoring_info.labels: | ||
return | ||
ptransform_label = monitoring_info.labels[ | ||
monitoring_infos.PTRANSFORM_LABEL] | ||
if not monitoring_infos.TAG_LABEL in monitoring_info.labels: | ||
return | ||
tag_label = monitoring_info.labels[monitoring_infos.TAG_LABEL] | ||
|
||
if not ptransform_label in self.process_bundle_descriptor.transforms: | ||
return | ||
if not tag_label in self.process_bundle_descriptor.transforms[ | ||
ptransform_label].outputs: | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Likewise this. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Actually, this can happen, and might be what's happening here. There is no PCollection for this tag, but the user outputted a value to this tag. It would make sense to record this output even if we didn't use it. This is another downside of attaching these counters to PCollections themselves rather than to PTransform outputs. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. unknown outputs should probably be reported another way |
||
return | ||
|
||
pcollection_name = ( | ||
self.process_bundle_descriptor.transforms[ptransform_label]. | ||
outputs[tag_label]) | ||
|
||
monitoring_info.labels[ | ||
monitoring_infos.PCOLLECTION_LABEL] = pcollection_name | ||
|
||
# Cleaning up labels that are not in specification. | ||
monitoring_info.labels.pop(monitoring_infos.PTRANSFORM_LABEL) | ||
monitoring_info.labels.pop(monitoring_infos.TAG_LABEL) | ||
|
||
for mi in infos_list: | ||
inject_pcollection(mi) | ||
|
||
return infos_list | ||
pcollection_ids = self.process_bundle_descriptor.transforms[ | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I believe the id order aligns with the receiver order since transform_consumers built above iterates the outputs map in the same order and this gets plumbed down through to Operation. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. In practice this might be OK (dicts have undefined, but I think when modified deterministic, iteration order), but seems rather brittle to me. Could we instead passed the tag -> pcollection_id mapping here? |
||
transform_id].outputs.values() | ||
all_monitoring_infos_dict.update( | ||
op.monitoring_infos(transform_id, pcollection_ids)) | ||
|
||
def _fix_output_tags_monitoring_info(self, transform_id, monitoring_info): | ||
# type: (str, metrics_pb2.MonitoringInfo) -> metrics_pb2.MonitoringInfo | ||
actual_output_tags = list( | ||
self.process_bundle_descriptor.transforms[transform_id].outputs.keys()) | ||
if ('TAG' in monitoring_info.labels and | ||
monitoring_info.labels['TAG'] == 'ONLY_OUTPUT'): | ||
if len(actual_output_tags) == 1: | ||
monitoring_info.labels['TAG'] = actual_output_tags[0] | ||
return monitoring_info | ||
return list(all_monitoring_infos_dict.values()) | ||
|
||
def shutdown(self): | ||
# type: () -> None | ||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -337,43 +337,48 @@ def add_receiver(self, operation, output_index=0): | |
"""Adds a receiver operation for the specified output.""" | ||
self.consumers[output_index].append(operation) | ||
|
||
def monitoring_infos(self, transform_id): | ||
# type: (str) -> Dict[FrozenSet, metrics_pb2.MonitoringInfo] | ||
def monitoring_infos(self, transform_id, pcollection_ids): | ||
# type: (str, list(str)) -> Dict[FrozenSet, metrics_pb2.MonitoringInfo] | ||
|
||
"""Returns the list of MonitoringInfos collected by this operation.""" | ||
all_monitoring_infos = self.execution_time_monitoring_infos(transform_id) | ||
all_monitoring_infos.update( | ||
self.pcollection_count_monitoring_infos(transform_id)) | ||
self.pcollection_count_monitoring_infos(pcollection_ids)) | ||
all_monitoring_infos.update(self.user_monitoring_infos(transform_id)) | ||
return all_monitoring_infos | ||
|
||
def pcollection_count_monitoring_infos(self, transform_id): | ||
def pcollection_count_monitoring_infos(self, pcollection_ids): | ||
"""Returns the element count MonitoringInfo collected by this operation.""" | ||
if len(self.receivers) == 1: | ||
# If there is exactly one output, we can unambiguously | ||
# fix its name later, which we do. | ||
# TODO(robertwb): Plumb the actual name here. | ||
if len(self.receivers) != len(pcollection_ids): | ||
raise RuntimeError( | ||
'Unexpected number of receivers for number of pcollections %s %s' % | ||
(self.receivers, pcollection_ids)) | ||
|
||
all_monitoring_infos = {} | ||
for i in range(len(self.receivers)): | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This will change if you use a mapping, but There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. +1 using zip |
||
receiver = self.receivers[i] | ||
pcollection_id = pcollection_ids[i] | ||
elem_count_mi = monitoring_infos.int64_counter( | ||
monitoring_infos.ELEMENT_COUNT_URN, | ||
self.receivers[0].opcounter.element_counter.value(), | ||
ptransform=transform_id, | ||
tag='ONLY_OUTPUT' if len(self.receivers) == 1 else str(None), | ||
receiver.opcounter.element_counter.value(), | ||
pcollection=pcollection_id, | ||
) | ||
|
||
(unused_mean, sum, count, min, max) = ( | ||
self.receivers[0].opcounter.mean_byte_counter.value()) | ||
receiver.opcounter.mean_byte_counter.value()) | ||
|
||
sampled_byte_count = monitoring_infos.int64_distribution( | ||
monitoring_infos.SAMPLED_BYTE_SIZE_URN, | ||
DistributionData(sum, count, min, max), | ||
ptransform=transform_id, | ||
tag='ONLY_OUTPUT' if len(self.receivers) == 1 else str(None), | ||
pcollection=pcollection_id, | ||
) | ||
return { | ||
monitoring_infos.to_key(elem_count_mi): elem_count_mi, | ||
monitoring_infos.to_key(sampled_byte_count): sampled_byte_count | ||
} | ||
return {} | ||
|
||
all_monitoring_infos[monitoring_infos.to_key( | ||
elem_count_mi)] = elem_count_mi | ||
all_monitoring_infos[monitoring_infos.to_key( | ||
sampled_byte_count)] = sampled_byte_count | ||
|
||
return all_monitoring_infos | ||
|
||
def user_monitoring_infos(self, transform_id): | ||
"""Returns the user MonitoringInfos collected by this operation.""" | ||
|
@@ -708,27 +713,6 @@ def reset(self): | |
self.user_state_context.reset() | ||
self.dofn_runner.bundle_finalizer_param.reset() | ||
|
||
def monitoring_infos(self, transform_id): | ||
# type: (str) -> Dict[FrozenSet, metrics_pb2.MonitoringInfo] | ||
infos = super(DoOperation, self).monitoring_infos(transform_id) | ||
if self.tagged_receivers: | ||
for tag, receiver in self.tagged_receivers.items(): | ||
mi = monitoring_infos.int64_counter( | ||
monitoring_infos.ELEMENT_COUNT_URN, | ||
receiver.opcounter.element_counter.value(), | ||
ptransform=transform_id, | ||
tag=str(tag)) | ||
infos[monitoring_infos.to_key(mi)] = mi | ||
(unused_mean, sum, count, min, max) = ( | ||
receiver.opcounter.mean_byte_counter.value()) | ||
sampled_byte_count = monitoring_infos.int64_distribution( | ||
monitoring_infos.SAMPLED_BYTE_SIZE_URN, | ||
DistributionData(sum, count, min, max), | ||
ptransform=transform_id, | ||
tag=str(tag)) | ||
infos[monitoring_infos.to_key(sampled_byte_count)] = sampled_byte_count | ||
return infos | ||
|
||
|
||
class SdfProcessSizedElements(DoOperation): | ||
def __init__(self, *args, **kwargs): | ||
|
@@ -777,7 +761,7 @@ def current_element_progress(self): | |
self.element_start_output_bytes) | ||
return None | ||
|
||
def monitoring_infos(self, transform_id): | ||
def monitoring_infos(self, transform_id, pcollection_ids): | ||
# type: (str) -> Dict[FrozenSet, metrics_pb2.MonitoringInfo] | ||
|
||
def encode_progress(value): | ||
|
@@ -787,7 +771,7 @@ def encode_progress(value): | |
|
||
with self.lock: | ||
infos = super(SdfProcessSizedElements, | ||
self).monitoring_infos(transform_id) | ||
self).monitoring_infos(transform_id, pcollection_ids) | ||
current_element_progress = self.current_element_progress() | ||
if current_element_progress: | ||
if current_element_progress.completed_work: | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This would be a bug, right?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
yeah