Support custom namespaces in MCF and observations importers - #661
Support custom namespaces in MCF and observations importers#661dwnoble wants to merge 2 commits into
Conversation
Up to standards ✅🟢 Issues
|
| Metric | Results |
|---|---|
| Complexity | 0 |
NEW Get contextual insights on your PRs based on Codacy's metrics, along with PR and Jira context, without leaving GitHub. Enable AI reviewer
TIP This summary will be updated as you push new changes.
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request refactors the data importers to support streaming and chunked processing of observations, reducing memory usage, and adds support for custom namespaces in entity resolution. Feedback on the changes highlights opportunities to optimize performance by deduplicating entity lists before resolution in both ObservationsImporter and EventsImporter to avoid redundant API requests, as well as a minor consistency improvement in logging paths.
| unresolved = [ | ||
| e for e in entities if e not in self._resolved_entities_cache | ||
| ] |
There was a problem hiding this comment.
Deduplicating the unresolved list of entities before calling _do_resolve is highly recommended. Since observations data often contains many duplicate entities (e.g., multiple dates or variables for the same entity), resolving duplicates externally can result in redundant network requests and API load. Deduplicating them locally improves performance and efficiency.
| unresolved = [ | |
| e for e in entities if e not in self._resolved_entities_cache | |
| ] | |
| unresolved = list(set( | |
| e for e in entities if e not in self._resolved_entities_cache | |
| )) |
| def _resolve(self, entities: list[str]) -> dict[str, str]: | ||
| lower_case_entity_name = self.entity_column_name.lower() | ||
|
|
||
| # Check if the entities can be resolved locally. | ||
| # If so, return them by prefixing the values as appropriate. | ||
| if lower_case_entity_name in constants.PRE_RESOLVED_INPUT_COLUMNS_TO_PREFIXES: | ||
| prefix = constants.PRE_RESOLVED_INPUT_COLUMNS_TO_PREFIXES[ | ||
| lower_case_entity_name] | ||
| return dict([(entity, f"{prefix}{entity}") for entity in entities]) | ||
|
|
||
| # Resolve entities externally. | ||
| property_name = constants.EXTERNALLY_RESOLVED_INPUT_COLUMNS_TO_PREFIXES.get( | ||
| lower_case_entity_name, constants.PROPERTY_DESCRIPTION) | ||
| return dc.resolve_entities(entities=entities, | ||
| entity_type=self.entity_type, | ||
| property_name=property_name) |
There was a problem hiding this comment.
Deduplicating the entities list before resolving them (either locally or externally) is highly recommended. Since events data can contain many duplicate entities, resolving duplicates externally can result in redundant network requests and API load. Deduplicating them locally improves performance and efficiency.
| def _resolve(self, entities: list[str]) -> dict[str, str]: | |
| lower_case_entity_name = self.entity_column_name.lower() | |
| # Check if the entities can be resolved locally. | |
| # If so, return them by prefixing the values as appropriate. | |
| if lower_case_entity_name in constants.PRE_RESOLVED_INPUT_COLUMNS_TO_PREFIXES: | |
| prefix = constants.PRE_RESOLVED_INPUT_COLUMNS_TO_PREFIXES[ | |
| lower_case_entity_name] | |
| return dict([(entity, f"{prefix}{entity}") for entity in entities]) | |
| # Resolve entities externally. | |
| property_name = constants.EXTERNALLY_RESOLVED_INPUT_COLUMNS_TO_PREFIXES.get( | |
| lower_case_entity_name, constants.PROPERTY_DESCRIPTION) | |
| return dc.resolve_entities(entities=entities, | |
| entity_type=self.entity_type, | |
| property_name=property_name) | |
| def _resolve(self, entities: list[str]) -> dict[str, str]: | |
| lower_case_entity_name = self.entity_column_name.lower() | |
| unique_entities = list(set(entities)) | |
| # Check if the entities can be resolved locally. | |
| # If so, return them by prefixing the values as appropriate. | |
| if lower_case_entity_name in constants.PRE_RESOLVED_INPUT_COLUMNS_TO_PREFIXES: | |
| prefix = constants.PRE_RESOLVED_INPUT_COLUMNS_TO_PREFIXES[ | |
| lower_case_entity_name] | |
| return dict([(entity, f"{prefix}{entity}") for entity in unique_entities]) | |
| # Resolve entities externally. | |
| property_name = constants.EXTERNALLY_RESOLVED_INPUT_COLUMNS_TO_PREFIXES.get( | |
| lower_case_entity_name, constants.PROPERTY_DESCRIPTION) | |
| return dc.resolve_entities(entities=unique_entities, | |
| entity_type=self.entity_type, | |
| property_name=property_name) |
| def _write_debug_csvs(self) -> None: | ||
| if self.debug_resolve_df is not None: | ||
| logging.info("Writing resolutions (for debugging) to: %s", | ||
| self.debug_resolve_file.path) |
There was a problem hiding this comment.
For consistency with EventsImporter and to provide more useful debugging information, log the full path of the debug resolve file (self.debug_resolve_file) instead of just the relative path (self.debug_resolve_file.path).
self.debug_resolve_file
5ad9973 to
8184c49
Compare
8184c49 to
a7dad39
Compare
| in ('location', 'observedNode', 'containedInPlace', 'containedIn', | ||
| 'populationType', 'measuredProperty', 'measurementDenominator', | ||
| 'populationGroup', 'constraintProperties', 'measurementMethod', | ||
| 'comparedNode', 'source', 'sourceLink')) |
There was a problem hiding this comment.
We got rid of sourceLink that was a mistake. It should just be source
| @@ -172,9 +173,12 @@ def _resolve_entities(self) -> None: | |||
| pre_resolved_entities = {} | |||
|
|
|||
| def remove_pre_resolved(entity: str) -> bool: | |||
There was a problem hiding this comment.
should we just make a shared helper called remove_pre_resolve?
| @@ -0,0 +1,7 @@ | |||
| { | |||
| "inputFiles": { | |||
There was a problem hiding this comment.
I think this is the legacy config.json format. Can you use the new one? (notice the input.csv is the key to the map
…o custom-namespace-importer-fix
2 participants
Add support for custom namespace subjects and objects in MCF and Observations importers, allowing custom namespace data sources to be successfully processed.
Highlights:
simple/kg_util/mcf_parser.pyandmcf_importer.pyto parse and maintain custom namespaces in the node properties.simple/stats/observations_importer.pyhandles namespaced identifiers during pre-resolution and resolving phases.custom_namespace_provenanceandcustom_namespace_observations).