make peek sample use consistent data#997
Conversation
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #997 +/- ##
==========================================
+ Coverage 65.78% 65.80% +0.02%
==========================================
Files 85 85
Lines 8204 8210 +6
==========================================
+ Hits 5397 5403 +6
Misses 2807 2807 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
drewoldag
left a comment
drewoldag
left a comment
There was a problem hiding this comment.
I have an open question about this PR that I would like to bottom out before approving.
| if hasattr(self, "pre_filter"): | ||
| sample = self.pre_filter([sample]) | ||
| if sample: | ||
| sample = sample[0] if isinstance(sample, list) else sample | ||
| self._buffered.append(sample) | ||
| # check if pre_process is defined and apply it to the sample | ||
| if hasattr(self, "pre_process"): | ||
| sample = self.pre_process([sample]) | ||
| return sample[0] if isinstance(sample, list) else sample |
There was a problem hiding this comment.
Before we merge this, let's think through if it makes sense to put this check here. I'm wondering it the same snippet of code might be better on the HyraxKafkaConsumer class in hyrax_alerts. i.e.
class HyraxKafkaConsumer(HyraxAlertsBaseConsumer, KafkaStreamDataset):
def __init__(self, config, data_location=None):
HyraxAlertsBaseConsumer.__init__(self, config=config, data_location=data_location)
KafkaStreamDataset.__init__(self, config=config, data_location=data_location)
def peek_sample(self):
sample = super.peek_sample()
# process `sample` with the new logic from this PR
return sampleThere was a problem hiding this comment.
I wonder only because I think KafkaStreamDataset can be completely unaware of pre filtering and processing and therefore (hopefully) more reusable.
But if I'm missing something obvious, please let me know :)
There was a problem hiding this comment.
my only question would be if that would not work nice with the wait/yield loop in the KafkaStreamDataset (i.e. what's the way that HyraxKafkaConsumer will know to continue the loop and wait for an alert if the alert returned out of super.peek_sample() get's pre-filtered and we need to wait for a sample again ?). It might require us to duplicate the loop in the overwritten version of the function which could get messy and lead to divergent functionality. let me give it a shot though !
| assert [s["object_id"] for s in batches[0]] == ["first", "second"] | ||
|
|
||
|
|
||
| def test_peek_sample_pre_filter(monkeypatch): |
There was a problem hiding this comment.
Same question as above, the tests looks reasonable, but I wonder if this belongs in hyrax_alerts.
2 participants
Change Description
in
hyrax_alerts, we define and allow the user to definepre_filterandpost_filterfunctions that might affect the data, butpeek_sample(used inhyrax_alerts.process_alerts()to do initial runtime setup) doesn't currently touch those functions, so we should check if those functions have been defined so thatpeek_samplecan maintain data consistency.Solution Description
check for the functions and then apply if them if available.
Note: I added the sample to
_bufferedbefore the sample is pre processed since we should assume that when it gets read again throughprocess_alertsas part of the main batch process that it'll be mixed in with data in the__iter__that hasn't been pre processed yet.Code Quality