feat(ingest): support knowledge links in business glossary

metadata-ingestion/examples/bootstrap_data/business_glossary.yml

@@ -7,6 +7,9 @@ url: "https://github.com/datahub-project/datahub/"
 nodes:
   - name: Classification
     description: A set of terms related to Data Classification
+    knowledge_links:
+      - label: Wiki link for classification
+        url: "https://en.wikipedia.org/wiki/Classification"
     terms:
       - name: Sensitive
         description: Sensitive Data
@@ -110,3 +113,6 @@ nodes:
         source_url: "https://spec.edmcouncil.org/fibo/ontology/FBC/ProductsAndServices/ClientsAndAccounts/Account"
         related_terms:
           - House.Kitchen
+        knowledge_links:
+          - url: "https://en.wikipedia.org/wiki/Spoon"
+            label: Wiki link

metadata-ingestion/src/datahub/ingestion/source/metadata/business_glossary.py

@@ -1,7 +1,10 @@
 import logging
+import pathlib
+import time
 from dataclasses import dataclass, field
 from typing import Any, Dict, Iterable, List, Optional, Union
 
+import pydantic
 from pydantic import validator
 from pydantic.fields import Field
 
@@ -14,6 +17,7 @@
     make_group_urn,
     make_user_urn,
 )
+from datahub.emitter.mcp import MetadataChangeProposalWrapper
 from datahub.ingestion.api.decorators import (  # SourceCapability,; capability,
     SupportStatus,
     config_class,
@@ -30,6 +34,8 @@
 # This needed to map path presents in inherits, contains, values, and related_terms to terms' optional id
 path_vs_id: Dict[str, Optional[str]] = {}
 
+URN_RESERVED_CHARS = ("%",)
+
 auditStamp = models.AuditStampClass(
     time=get_sys_time(), actor="urn:li:corpUser:restEmitter"
 )
@@ -40,6 +46,11 @@ class Owners(ConfigModel):
     groups: Optional[List[str]]
 
 
+class KnowledgeCard(ConfigModel):
+    url: Optional[str]
+    label: Optional[str]
+
+
 class GlossaryTermConfig(ConfigModel):
     id: Optional[str]
     name: str
@@ -53,6 +64,7 @@ class GlossaryTermConfig(ConfigModel):
     values: Optional[List[str]]
     related_terms: Optional[List[str]]
     custom_properties: Optional[Dict[str, str]]
+    knowledge_links: Optional[List[KnowledgeCard]]
 
 
 class GlossaryNodeConfig(ConfigModel):
@@ -62,6 +74,7 @@ class GlossaryNodeConfig(ConfigModel):
     owners: Optional[Owners]
     terms: Optional[List[GlossaryTermConfig]]
     nodes: Optional[List["GlossaryNodeConfig"]]
+    knowledge_links: Optional[List[KnowledgeCard]]
 
 
 GlossaryNodeConfig.update_forward_refs()
@@ -77,7 +90,9 @@ class DefaultConfig(ConfigModel):
 
 
 class BusinessGlossarySourceConfig(ConfigModel):
-    file: str = Field(description="Path to business glossary file to ingest.")
+    file: pydantic.FilePath = Field(
+        description="Path to business glossary file to ingest."
+    )
     enable_auto_id: bool = Field(
         description="Generate id field from GlossaryNode and GlossaryTerm's name field",
         default=False,
@@ -101,6 +116,11 @@ def create_id(path: List[str], default_id: Optional[str], enable_auto_id: bool)
         return default_id  # No need to create id from path as default_id is provided
 
     id_: str = ".".join(path)
+
+    for c in id_:
+        if c in URN_RESERVED_CHARS:
+            enable_auto_id = True
+
     if enable_auto_id:
         id_ = datahub_guid({"path": id_})
     return id_
@@ -153,14 +173,13 @@ def get_owners(owners: Owners) -> models.OwnershipClass:
 
 def get_mces(
     glossary: BusinessGlossaryConfig, ingestion_config: BusinessGlossarySourceConfig
-) -> List[models.MetadataChangeEventClass]:
-    events: List[models.MetadataChangeEventClass] = []
+) -> Iterable[Union[MetadataChangeProposalWrapper, models.MetadataChangeEventClass]]:
     path: List[str] = []
     root_owners = get_owners(glossary.owners)
 
     if glossary.nodes:
         for node in glossary.nodes:
-            events += get_mces_from_node(
+            yield from get_mces_from_node(
                 node,
                 path + [node.name],
                 parentNode=None,
@@ -171,7 +190,7 @@ def get_mces(
 
     if glossary.terms:
         for term in glossary.terms:
-            events += get_mces_from_term(
+            yield from get_mces_from_term(
                 term,
                 path + [term.name],
                 parentNode=None,
@@ -180,21 +199,47 @@ def get_mces(
                 ingestion_config=ingestion_config,
             )
 
-    return events
-
 
 def get_mce_from_snapshot(snapshot: Any) -> models.MetadataChangeEventClass:
     return models.MetadataChangeEventClass(proposedSnapshot=snapshot)
 
 
+def make_institutional_memory_mcp(
+    urn: str, knowledge_cards: List[KnowledgeCard]
+) -> Optional[MetadataChangeProposalWrapper]:
+    elements: List[models.InstitutionalMemoryMetadataClass] = []
+
+    for knowledge_card in knowledge_cards:
+        if knowledge_card.label and knowledge_card.url:
+            elements.append(
+                models.InstitutionalMemoryMetadataClass(
+                    url=knowledge_card.url,
+                    description=knowledge_card.label,
+                    createStamp=models.AuditStampClass(
+                        time=int(time.time() * 1000.0),
+                        actor="urn:li:corpuser:datahub",
+                        message="ingestion bot",
+                    ),
+                )
+            )
+
+    if elements:
+        return MetadataChangeProposalWrapper(
+            entityUrn=urn,
+            aspect=models.InstitutionalMemoryClass(elements=elements),
+        )
+
+    return None
+
+
 def get_mces_from_node(
     glossaryNode: GlossaryNodeConfig,
     path: List[str],
     parentNode: Optional[str],
     parentOwners: models.OwnershipClass,
     defaults: DefaultConfig,
     ingestion_config: BusinessGlossarySourceConfig,
-) -> List[models.MetadataChangeEventClass]:
+) -> Iterable[Union[MetadataChangeProposalWrapper, models.MetadataChangeEventClass]]:
     node_urn = make_glossary_node_urn(
         path, glossaryNode.id, ingestion_config.enable_auto_id
     )
@@ -212,10 +257,18 @@ def get_mces_from_node(
         urn=node_urn,
         aspects=[node_info, node_owners, valid_status],
     )
-    mces = [get_mce_from_snapshot(node_snapshot)]
+    yield get_mce_from_snapshot(node_snapshot)
+
+    if glossaryNode.knowledge_links is not None:
+        mcp: Optional[MetadataChangeProposalWrapper] = make_institutional_memory_mcp(
+            node_urn, glossaryNode.knowledge_links
+        )
+        if mcp is not None:
+            yield mcp
+
     if glossaryNode.nodes:
         for node in glossaryNode.nodes:
-            mces += get_mces_from_node(
+            yield from get_mces_from_node(
                 node,
                 path + [node.name],
                 parentNode=node_urn,
@@ -226,15 +279,14 @@ def get_mces_from_node(
 
     if glossaryNode.terms:
         for term in glossaryNode.terms:
-            mces += get_mces_from_term(
+            yield from get_mces_from_term(
                 glossaryTerm=term,
                 path=path + [term.name],
                 parentNode=node_urn,
                 parentOwnership=node_owners,
                 defaults=defaults,
                 ingestion_config=ingestion_config,
             )
-    return mces
 
 
 def get_mces_from_term(
@@ -244,7 +296,7 @@ def get_mces_from_term(
     parentOwnership: models.OwnershipClass,
     defaults: DefaultConfig,
     ingestion_config: BusinessGlossarySourceConfig,
-) -> List[models.MetadataChangeEventClass]:
+) -> Iterable[Union[models.MetadataChangeEventClass, MetadataChangeProposalWrapper]]:
     term_urn = make_glossary_term_urn(
         path, glossaryTerm.id, ingestion_config.enable_auto_id
     )
@@ -338,14 +390,18 @@ def get_mces_from_term(
         ownership = get_owners(glossaryTerm.owners)
     aspects.append(ownership)
 
-    term_browse = models.BrowsePathsClass(paths=["/" + "/".join(path)])
-    aspects.append(term_browse)
-
     term_snapshot: models.GlossaryTermSnapshotClass = models.GlossaryTermSnapshotClass(
         urn=term_urn,
         aspects=aspects,
     )
-    return [get_mce_from_snapshot(term_snapshot)]
+    yield get_mce_from_snapshot(term_snapshot)
+
+    if glossaryTerm.knowledge_links:
+        mcp: Optional[MetadataChangeProposalWrapper] = make_institutional_memory_mcp(
+            term_urn, glossaryTerm.knowledge_links
+        )
+        if mcp is not None:
+            yield mcp
 
 
 def populate_path_vs_id(glossary: BusinessGlossaryConfig) -> None:
@@ -388,18 +444,25 @@ def create(cls, config_dict, ctx):
         config = BusinessGlossarySourceConfig.parse_obj(config_dict)
         return cls(ctx, config)
 
-    def load_glossary_config(self, file_name: str) -> BusinessGlossaryConfig:
+    def load_glossary_config(self, file_name: pathlib.Path) -> BusinessGlossaryConfig:
         config = load_config_file(file_name)
         glossary_cfg = BusinessGlossaryConfig.parse_obj(config)
         return glossary_cfg
 
     def get_workunits(self) -> Iterable[Union[MetadataWorkUnit, UsageStatsWorkUnit]]:
         glossary_config = self.load_glossary_config(self.config.file)
         populate_path_vs_id(glossary_config)
-        for mce in get_mces(glossary_config, ingestion_config=self.config):
-            wu = MetadataWorkUnit(f"{mce.proposedSnapshot.urn}", mce=mce)
-            self.report.report_workunit(wu)
-            yield wu
+        for event in get_mces(glossary_config, ingestion_config=self.config):
+            if isinstance(event, models.MetadataChangeEventClass):
+                wu = MetadataWorkUnit(f"{event.proposedSnapshot.urn}", mce=event)
+                self.report.report_workunit(wu)
+                yield wu
+            elif isinstance(event, MetadataChangeProposalWrapper):
+                wu = MetadataWorkUnit(
+                    id=f"{event.entityType}-{event.aspectName}-{event.entityUrn}",
+                    mcp=event,
+                )
+                yield wu
 
     def get_report(self):
         return self.report

metadata-ingestion/tests/integration/business-glossary/business_glossary.yml

@@ -0,0 +1,72 @@
+version: 1
+source: DataHub
+owners:
+  users:
+    - mjames
+url: "https://github.com/datahub-project/datahub/"
+nodes:
+  - name: Classification
+    description: A set of terms related to Data Classification
+    knowledge_links:
+      - label: Wiki link for classification
+        url: "https://en.wikipedia.org/wiki/Classification"
+    terms:
+      - name: Sensitive
+        description: Sensitive Data
+        custom_properties:
+          is_confidential: false
+        knowledge_links:
+          - label: Google Link
+            url: "https://www.google.com"
+      - name: Confidential
+        description: Confidential Data
+        custom_properties:
+          is_confidential: true
+      - name: Highly Confidential
+        description: Highly Confidential Data
+        custom_properties:
+          is_confidential: true
+  - name: Personal Information
+    description: All terms related to personal information
+    owners:
+      users:
+        - mjames
+    terms:
+      - name: Email
+        description: An individual's email address
+        inherits:
+          - Classification.Confidential
+        owners:
+          groups:
+            - Trust and Safety
+      - name: Address
+        description: A physical address
+      - name: Gender
+        description: The gender identity of the individual
+        inherits:
+          - Classification.Sensitive
+  - name: Clients And Accounts
+    description: Provides basic concepts such as account, account holder, account provider, relationship manager that are commonly used by financial services providers to describe customers and to determine counterparty identities
+    owners:
+      groups:
+        - finance
+    terms:
+      - name: Account
+        description: Container for records associated with a business arrangement for regular transactions and services
+        term_source: "EXTERNAL"
+        source_ref: FIBO
+        source_url: "https://spec.edmcouncil.org/fibo/ontology/FBC/ProductsAndServices/ClientsAndAccounts/Account"
+        inherits:
+          - Classification.Highly Confidential
+        contains:
+          - Clients And Accounts.Balance
+      - name: Balance
+        description: Amount of money available or owed
+        term_source: "EXTERNAL"
+        source_ref: FIBO
+        source_url: "https://spec.edmcouncil.org/fibo/ontology/FBC/ProductsAndServices/ClientsAndAccounts/Balance"
+  - name: KPIs
+    description: Common Business KPIs
+    terms:
+      - name: CSAT %
+        description: Customer Satisfaction Score