diff --git a/datahub-web-react/src/app/ingest/source/builder/constants.ts b/datahub-web-react/src/app/ingest/source/builder/constants.ts index 457ea50308d558..b1b2a1ea987d2d 100644 --- a/datahub-web-react/src/app/ingest/source/builder/constants.ts +++ b/datahub-web-react/src/app/ingest/source/builder/constants.ts @@ -36,6 +36,7 @@ import qlikLogo from '@images/qliklogo.png'; import redshiftLogo from '@images/redshiftlogo.png'; import sacLogo from '@images/saclogo.svg'; import sigmaLogo from '@images/sigmalogo.png'; +import snaplogic from '@images/snaplogic.png'; import snowflakeLogo from '@images/snowflakelogo.png'; import supersetLogo from '@images/supersetlogo.png'; import tableauLogo from '@images/tableaulogo.png'; @@ -149,6 +150,8 @@ export const NEO4J = 'neo4j'; export const NEO4J_URN = `urn:li:dataPlatform:${NEO4J}`; export const VERTEX_AI = 'vertexai'; export const VERTEXAI_URN = `urn:li:dataPlatform:${VERTEX_AI}`; +export const SNAPLOGIC = 'snaplogic'; +export const SNAPLOGIC_URN = `urn:li:dataPlatform:${SNAPLOGIC}`; export const PLATFORM_URN_TO_LOGO = { [ATHENA_URN]: athenaLogo, @@ -196,6 +199,7 @@ export const PLATFORM_URN_TO_LOGO = { [DATAHUB_URN]: datahubLogo, [NEO4J_URN]: neo4j, [VERTEXAI_URN]: vertexAI, + [SNAPLOGIC_URN]: snaplogic, }; export const SOURCE_TO_PLATFORM_URN = { diff --git a/datahub-web-react/src/app/ingest/source/builder/sources.json b/datahub-web-react/src/app/ingest/source/builder/sources.json index 624f19f47b85b5..db31db73da9c7a 100644 --- a/datahub-web-react/src/app/ingest/source/builder/sources.json +++ b/datahub-web-react/src/app/ingest/source/builder/sources.json @@ -15,6 +15,14 @@ "docsUrl": "https://docs.datahub.com/docs/quick-ingestion-guides/redshift/overview", "recipe": "source: \n type: redshift\n config:\n # Coordinates\n host_port: # Your Redshift host and post, e.g. example.something.us-west-2.redshift.amazonaws.com:5439\n database: # Your Redshift database, e.g. SampleDatabase\n\n # Credentials\n # Add secret in Secrets Tab with relevant names for each variable\n username: null # Your Redshift username, e.g. admin\n\n table_lineage_mode: stl_scan_based\n include_table_lineage: true\n include_tables: true\n include_views: true\n profiling:\n enabled: true\n profile_table_level_only: true\n stateful_ingestion:\n enabled: true" }, + { + "urn": "urn:li:dataPlatform:snaplogic", + "name": "snaplogic", + "displayName": "Snaplogic", + "description": "Import lineage from Snaplogic.", + "docsUrl": "https://docs.datahub.com/docs/quick-ingestion-guides/snaplogic/overview", + "recipe": "source:\n type: snaplogic\n config:\n username: # username\n password: # password\n base_url: https://elastic.snaplogic.com\n org_name: # Organization name from Snaplogic instance\n stateful_ingestion:\n enabled: True\n remove_stale_metadata: False\n" + }, { "urn": "urn:li:dataPlatform:snowflake", "name": "snowflake", diff --git a/datahub-web-react/src/app/ingestV2/source/builder/sources.json b/datahub-web-react/src/app/ingestV2/source/builder/sources.json index 253ce8219456f1..8eea4a4a0f759b 100644 --- a/datahub-web-react/src/app/ingestV2/source/builder/sources.json +++ b/datahub-web-react/src/app/ingestV2/source/builder/sources.json @@ -15,6 +15,14 @@ "docsUrl": "https://docs.datahub.com/docs/quick-ingestion-guides/redshift/overview", "recipe": "source: \n type: redshift\n config:\n # Coordinates\n host_port: # Your Redshift host and post, e.g. example.something.us-west-2.redshift.amazonaws.com:5439\n database: # Your Redshift database, e.g. SampleDatabase\n\n # Credentials\n # Add secret in Secrets Tab with relevant names for each variable\n username: null # Your Redshift username, e.g. admin\n\n table_lineage_mode: stl_scan_based\n include_table_lineage: true\n include_tables: true\n include_views: true\n profiling:\n enabled: true\n profile_table_level_only: true\n stateful_ingestion:\n enabled: true" }, + { + "urn": "urn:li:dataPlatform:snaplogic", + "name": "snaplogic", + "displayName": "Snaplogic", + "description": "Import lineage from Snaplogic.", + "docsUrl": "https://docs.datahub.com/docs/quick-ingestion-guides/snaplogic/overview", + "recipe": "source:\n type: snaplogic\n config:\n username: # username\n password: # password\n base_url: https://elastic.snaplogic.com\n org_name: # Organization name from Snaplogic instance\n stateful_ingestion:\n enabled: True\n remove_stale_metadata: False\n" + }, { "urn": "urn:li:dataPlatform:snowflake", "name": "snowflake", diff --git a/datahub-web-react/src/images/snaplogic.png b/datahub-web-react/src/images/snaplogic.png new file mode 100644 index 00000000000000..8e786e3b9108c7 Binary files /dev/null and b/datahub-web-react/src/images/snaplogic.png differ diff --git a/datahub-web-react/src/images/snaplogic.svg b/datahub-web-react/src/images/snaplogic.svg new file mode 100644 index 00000000000000..6af42654e82aed --- /dev/null +++ b/datahub-web-react/src/images/snaplogic.svg @@ -0,0 +1,74 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/metadata-ingestion/docs/sources/snaplogic/snaplogic_pre.md b/metadata-ingestion/docs/sources/snaplogic/snaplogic_pre.md new file mode 100644 index 00000000000000..b55111366cad04 --- /dev/null +++ b/metadata-ingestion/docs/sources/snaplogic/snaplogic_pre.md @@ -0,0 +1,72 @@ +## Integration Details + + + + +This integration extracts data lineage information from the public SnapLogic Lineage API and ingests it into DataHub. It enables visibility into how data flows through SnapLogic pipelines by capturing metadata directly from the source API. This allows users to track data transformations and dependencies across their data ecosystem, enhancing observability, governance, and impact analysis within DataHub. + +### Concept Mapping + + + + +This ingestion source maps the following Source System Concepts to DataHub Concepts: + + + +| Source Concept | DataHub Concept | Notes | +| -------------- | ------------------------------------------------------------------ | --------------------------------------------------------------------------------------------------------------------------------------- | +| Snap-pack | [Data Platform](docs/generated/metamodel/entities/dataPlatform.md) | Snap-packs are mapped to Data Platforms, either directly (e.g., Snowflake) or dynamically based on connection details (e.g., JDBC URL). | +| Table/Dataset | [Dataset](docs/generated/metamodel/entities/dataset.md) | May be differernt. It depends on a snap type. For sql databases it's table. For kafka it's topic, etc | +| Snap | [Data Job](docs/generated/metamodel/entities/dataJob.md) | | +| Pipeline | [Data Flow](docs/generated/metamodel/entities/dataFlow.md) | | + +## Metadata Ingestion Quickstart + +### Prerequisites + +In order to ingest lineage from snaplogic, you will need valid snaplogic credentials with access to the SnapLogic Lineage API. + +### Install the Plugin(s) + +Run the following commands to install the relevant plugin(s): + +`pip install 'acryl-datahub[snaplogic]'` + +### Configure the Ingestion Recipe(s) + +Use the following recipe(s) to get started with ingestion. + +#### `'acryl-datahub[snaplogic]'` + +```yml +pipeline_name: +source: + type: snaplogic + config: + username: + password: + base_url: https://elastic.snaplogic.com + org_name: + stateful_ingestion: + enabled: True + remove_stale_metadata: False +``` + +
+ View All Recipe Configuartion Options + +| Field | Required | Default | Description | +| ----------------------------- | :------: | :---------------------------: | --------------------------------------------------------------- | +| `username` | ✅ | | SnapLogic account login | +| `password` | ✅ | | SnapLogic account password. | +| `base_url` | ✅ | https://elastic.snaplogic.com | Snaplogic url | +| `org_name` | ✅ | | Organisation name in snaplogic platform | +| `namespace_mapping` | ❌ | | Namespace mapping. Used to map namespaces to platform instances | +| `case_insensitive_namespaces` | ❌ | | List of case insensitive namespaces | + +
+ +## Troubleshooting + +### [Common Issue] diff --git a/metadata-ingestion/docs/sources/snaplogic/snaplogic_recipe.yml b/metadata-ingestion/docs/sources/snaplogic/snaplogic_recipe.yml new file mode 100644 index 00000000000000..83f8acb79fe3e1 --- /dev/null +++ b/metadata-ingestion/docs/sources/snaplogic/snaplogic_recipe.yml @@ -0,0 +1,15 @@ +pipeline_name: "snaplogic_incremental_ingestion" +source: + type: snaplogic + config: + username: example@snaplogic.com + password: password + base_url: https://elastic.snaplogic.com + org_name: "ExampleOrg" + namespace_mapping: + snowflake://snaplogic: snaplogic + case_insensitive_namespaces: + - snowflake://snaplogic + stateful_ingestion: + enabled: True + remove_stale_metadata: False diff --git a/metadata-ingestion/setup.py b/metadata-ingestion/setup.py index 27aaf39260a9d8..d1c52ba1f5499d 100644 --- a/metadata-ingestion/setup.py +++ b/metadata-ingestion/setup.py @@ -569,6 +569,7 @@ # databricks is alias for unity-catalog and needs to be kept in sync "databricks": databricks | sql_common, "fivetran": snowflake_common | bigquery_common | sqlalchemy_lib | sqlglot_lib, + "snaplogic": set(), "qlik-sense": sqlglot_lib | {"requests", "websocket-client"}, "sigma": sqlglot_lib | {"requests"}, "sac": sac, @@ -700,6 +701,7 @@ "redshift", "s3", "snowflake", + "snaplogic", "slack", "tableau", "teradata", @@ -842,6 +844,7 @@ "gcs = datahub.ingestion.source.gcs.gcs_source:GCSSource", "sql-queries = datahub.ingestion.source.sql_queries:SqlQueriesSource", "fivetran = datahub.ingestion.source.fivetran.fivetran:FivetranSource", + "snaplogic = datahub.ingestion.source.snaplogic.snaplogic:SnaplogicSource", "qlik-sense = datahub.ingestion.source.qlik_sense.qlik_sense:QlikSenseSource", "sigma = datahub.ingestion.source.sigma.sigma:SigmaSource", "sac = datahub.ingestion.source.sac.sac:SACSource", diff --git a/metadata-ingestion/src/datahub/ingestion/autogenerated/capability_summary.json b/metadata-ingestion/src/datahub/ingestion/autogenerated/capability_summary.json index 82596cfbb921e2..2a39053492fdf1 100644 --- a/metadata-ingestion/src/datahub/ingestion/autogenerated/capability_summary.json +++ b/metadata-ingestion/src/datahub/ingestion/autogenerated/capability_summary.json @@ -2922,6 +2922,38 @@ "platform_name": "Slack", "support_status": "TESTING" }, + "snaplogic": { + "capabilities": [ + { + "capability": "LINEAGE_FINE", + "description": "Enabled by default", + "subtype_modifier": null, + "supported": true + }, + { + "capability": "DELETION_DETECTION", + "description": "Not supported yet", + "subtype_modifier": null, + "supported": false + }, + { + "capability": "PLATFORM_INSTANCE", + "description": "Snaplogic does not support platform instances", + "subtype_modifier": null, + "supported": false + }, + { + "capability": "LINEAGE_COARSE", + "description": "Enabled by default", + "subtype_modifier": null, + "supported": true + } + ], + "classname": "datahub.ingestion.source.snaplogic.snaplogic.SnaplogicSource", + "platform_id": "snaplogic", + "platform_name": "Snaplogic", + "support_status": "TESTING" + }, "snowflake": { "capabilities": [ { @@ -3565,4 +3597,4 @@ "support_status": "CERTIFIED" } } -} \ No newline at end of file +} diff --git a/metadata-ingestion/src/datahub/ingestion/source/snaplogic/__init__.py b/metadata-ingestion/src/datahub/ingestion/source/snaplogic/__init__.py new file mode 100644 index 00000000000000..e69de29bb2d1d6 diff --git a/metadata-ingestion/src/datahub/ingestion/source/snaplogic/snaplogic.py b/metadata-ingestion/src/datahub/ingestion/source/snaplogic/snaplogic.py new file mode 100644 index 00000000000000..47c06e7ae27bb6 --- /dev/null +++ b/metadata-ingestion/src/datahub/ingestion/source/snaplogic/snaplogic.py @@ -0,0 +1,349 @@ +from typing import Iterable, List, Optional + +from datahub.emitter.mce_builder import ( + make_data_flow_urn, + make_data_job_urn, + make_data_platform_urn, + make_dataset_urn_with_platform_instance, + make_schema_field_urn, +) +from datahub.emitter.mcp import MetadataChangeProposalWrapper +from datahub.ingestion.api.common import PipelineContext +from datahub.ingestion.api.decorators import ( + SupportStatus, + capability, + config_class, + platform_name, + support_status, +) +from datahub.ingestion.api.source import ( + MetadataWorkUnitProcessor, + SourceCapability, + SourceReport, +) +from datahub.ingestion.api.workunit import MetadataWorkUnit +from datahub.ingestion.graph.client import DataHubGraph +from datahub.ingestion.source.snaplogic.snaplogic_config import SnaplogicConfig +from datahub.ingestion.source.snaplogic.snaplogic_lineage_extractor import ( + SnaplogicLineageExtractor, +) +from datahub.ingestion.source.snaplogic.snaplogic_parser import ( + ColumnMapping, + Dataset, + SnapLogicParser, +) +from datahub.ingestion.source.state.redundant_run_skip_handler import ( + RedundantLineageRunSkipHandler, +) +from datahub.ingestion.source.state.stale_entity_removal_handler import ( + StaleEntityRemovalHandler, + StaleEntityRemovalSourceReport, +) +from datahub.ingestion.source.state.stateful_ingestion_base import ( + StatefulIngestionSourceBase, +) +from datahub.metadata.schema_classes import ( + DataFlowInfoClass, + DataJobInfoClass, + DataJobInputOutputClass, + DatasetPropertiesClass, + FineGrainedLineageClass, + FineGrainedLineageDownstreamTypeClass, + OtherSchemaClass, + SchemaFieldClass, + SchemaFieldDataTypeClass, + SchemaMetadataClass, + StringTypeClass, +) + + +@platform_name("Snaplogic") +@config_class(SnaplogicConfig) +@support_status(SupportStatus.TESTING) +@capability( + SourceCapability.PLATFORM_INSTANCE, + "Snaplogic does not support platform instances", + supported=False, +) +@capability(SourceCapability.LINEAGE_COARSE, "Enabled by default") +@capability(SourceCapability.LINEAGE_FINE, "Enabled by default") +@capability(SourceCapability.DELETION_DETECTION, "Not supported yet", supported=False) +class SnaplogicSource(StatefulIngestionSourceBase): + """ + A source plugin for ingesting lineage and metadata from Snaplogic. + """ + + def __init__(self, config: SnaplogicConfig, ctx: PipelineContext): + super().__init__(config, ctx) + self.config = config + self.report = StaleEntityRemovalSourceReport() + self.graph: Optional[DataHubGraph] = ctx.graph + self.snaplogic_parser = SnapLogicParser( + config.case_insensitive_namespaces, self.config.namespace_mapping + ) + self.redundant_lineage_run_skip_handler: Optional[ + RedundantLineageRunSkipHandler + ] = None + if self.config.enable_stateful_lineage_ingestion: + self.redundant_lineage_run_skip_handler = RedundantLineageRunSkipHandler( + source=self, + config=self.config, + pipeline_name=ctx.pipeline_name, + run_id=ctx.run_id, + ) + self.snaplogic_lineage_extractor = SnaplogicLineageExtractor( + config=config, + redundant_run_skip_handler=self.redundant_lineage_run_skip_handler, + report=self.report, + ) + + def get_workunits_internal(self) -> Iterable[MetadataWorkUnit]: + try: + self.report.info( + message="Starting lineage ingestion from Snaplogic", + title="Lineage Ingestion", + ) + + records_processed = 0 + for lineage in self.snaplogic_lineage_extractor.get_lineages(): + try: + for workunit in self._process_lineage_record(lineage): + yield workunit + records_processed += 1 + + if records_processed % 20 == 0: + self.report.info( + message=f"Processed {records_processed} lineage records", + title="Lineage Ingestion Progress", + ) + except Exception as e: + self.report.report_failure( + message="Failed to process lineage record", + context=str(lineage), + exc=e, + ) + self.report.info( + message=f"Completed processing {records_processed} lineage records", + title="Lineage Ingestion Complete", + ) + self.snaplogic_lineage_extractor.report_status("lineage_ingestion", True) + self.snaplogic_lineage_extractor.update_stats() + except Exception as e: + self.report.report_failure(message="Failed to fetch lineages", exc=e) + self.snaplogic_lineage_extractor.report_status("lineage_ingestion", False) + + def get_workunit_processors(self) -> List[Optional[MetadataWorkUnitProcessor]]: + return [ + *super().get_workunit_processors(), + StaleEntityRemovalHandler.create( + self, self.config, self.ctx + ).workunit_processor, + ] + + def _process_lineage_record(self, lineage: dict) -> Iterable[MetadataWorkUnit]: + """Process a lineage record to create pipeline and task workunits with relationships.""" + producer = lineage.get("producer") + if not producer: + return + pipeline_snode_id = producer.split("#pipe_snode=")[1] + if not pipeline_snode_id: + return + datasets = self.snaplogic_parser.extract_datasets_from_lineage(lineage) + pipeline = self.snaplogic_parser.extract_pipeline_from_lineage(lineage) + task = self.snaplogic_parser.extract_task_from_lineage(lineage) + columns_mapping = self.snaplogic_parser.extract_columns_mapping_from_lineage( + lineage + ) + + # Create pipeline MCP + for pipeline_workunit in self.create_pipeline_mcp( + name=pipeline.name, + pipeline_snode_id=pipeline.id, + namespace=pipeline.namespace, + ): + self.report.report_workunit(pipeline_workunit) + yield pipeline_workunit + + # Create dataset MCP + for dataset in datasets: + for dataset_workunit in self.create_dataset_mcp( + dataset_name=dataset.name, + dataset_display_name=dataset.display_name, + fields=dataset.fields, + platform=dataset.platform, + platform_instance=dataset.platform_instance, + ): + self.report.report_workunit(dataset_workunit) + yield dataset_workunit + + # Create task MCP + for task_workunit in self.create_task_mcp( + name=task.name, + task_id=task.id, + namespace=task.namespace, + pipeline_snode_id=pipeline_snode_id, + input_datasets=[dataset for dataset in datasets if dataset.type == "INPUT"], + output_datasets=[ + dataset for dataset in datasets if dataset.type == "OUTPUT" + ], + columns_mapping=columns_mapping, + ): + self.report.report_workunit(task_workunit) + yield task_workunit + + def create_task_mcp( + self, + task_id: str, + name: str, + namespace: str, + pipeline_snode_id: str, + input_datasets: list[Dataset], + output_datasets: list[Dataset], + columns_mapping: list[ColumnMapping], + ) -> Iterable[MetadataWorkUnit]: + """Create MCPs for a task (snap) including metadata and lineage.""" + job_urn = make_data_job_urn( + orchestrator=namespace, + flow_id=pipeline_snode_id, + job_id=task_id, + cluster="PROD", + ) + yield MetadataChangeProposalWrapper( + entityUrn=job_urn, + aspect=DataJobInfoClass( + name=name, + description="", + externalUrl=f"{self.config.base_url}/sl/designer.html?v=21818#pipe_snode={pipeline_snode_id}", + type="SNAPLOGIC_SNAP", + ), + ).as_workunit() + + # Helper functions + def dataset_urn(d: Dataset) -> str: + return make_dataset_urn_with_platform_instance( + d.platform, d.name, d.platform_instance + ) + + def field_urn(d, f): + return make_schema_field_urn(dataset_urn(d), f["name"]) + + # Emit lineage + yield MetadataChangeProposalWrapper( + entityUrn=job_urn, + aspect=DataJobInputOutputClass( + inputDatasets=[dataset_urn(d) for d in input_datasets], + outputDatasets=[dataset_urn(d) for d in output_datasets], + inputDatasetFields=[ + field_urn(d, f) for d in input_datasets for f in d.fields + ], + outputDatasetFields=[ + field_urn(d, f) for d in output_datasets for f in d.fields + ], + fineGrainedLineages=[ + FineGrainedLineageClass( + upstreamType=FineGrainedLineageDownstreamTypeClass.FIELD_SET, + upstreams=[ + make_schema_field_urn( + make_dataset_urn_with_platform_instance( + cl.input_dataset.platform, + cl.input_dataset.name, + cl.input_dataset.platform_instance, + cl.input_dataset.env, + ), + cl.input_field, + ) + ], + downstreamType=FineGrainedLineageDownstreamTypeClass.FIELD_SET, + downstreams=[ + make_schema_field_urn( + make_dataset_urn_with_platform_instance( + cl.output_dataset.platform, + cl.output_dataset.name, + cl.output_dataset.platform_instance, + cl.output_dataset.env, + ), + cl.output_field, + ) + ], + ) + for cl in columns_mapping + ], + ), + ).as_workunit() + + def create_dataset_mcp( + self, + dataset_name: str, + dataset_display_name: str, + fields: list[dict], + platform: str = "snaplogic", + env: str = "PROD", + platform_instance: Optional[str] = None, + ) -> Iterable[MetadataWorkUnit]: + dataset_urn = make_dataset_urn_with_platform_instance( + platform=platform, + name=dataset_name, + env=env, + platform_instance=platform_instance, + ) + # If we have datasets already in the datahub we don't need to create it. + # We only create datasets for other platforms if they don't exist in the datahub. + if platform != "snaplogic" and self.graph and self.graph.exists(dataset_urn): + return + + dataset_properties = DatasetPropertiesClass( + name=dataset_display_name, + qualifiedName=dataset_name, + ) + schema_fields = [ + SchemaFieldClass( + fieldPath=field["name"], + type=SchemaFieldDataTypeClass(StringTypeClass()), + nativeDataType=field.get("type", "Varchar"), + ) + for field in fields + ] + schema_metadata = SchemaMetadataClass( + schemaName=dataset_name, + platform=make_data_platform_urn(platform), + version=0, + hash="", + platformSchema=OtherSchemaClass(rawSchema=""), + fields=schema_fields, + ) + + yield MetadataChangeProposalWrapper( + entityUrn=dataset_urn, aspect=dataset_properties + ).as_workunit() + + yield MetadataChangeProposalWrapper( + entityUrn=dataset_urn, aspect=schema_metadata + ).as_workunit() + + def create_pipeline_mcp( + self, name: str, namespace: str, pipeline_snode_id: str + ) -> Iterable[MetadataWorkUnit]: + flow_urn = make_data_flow_urn( + orchestrator=namespace, flow_id=pipeline_snode_id, cluster="PROD" + ) + + yield MetadataChangeProposalWrapper( + entityUrn=flow_urn, + aspect=DataFlowInfoClass( + name=name, + description="", + externalUrl=f"{self.config.base_url}/sl/designer.html?v=21818#pipe_snode={pipeline_snode_id}", + ), + ).as_workunit() + + def get_report(self) -> SourceReport: + return self.report + + def close(self) -> None: + super().close() + StatefulIngestionSourceBase.close(self) + + @classmethod + def create(cls, config_dict: dict, ctx: PipelineContext) -> "SnaplogicSource": + config = SnaplogicConfig.parse_obj(config_dict) + return cls(config, ctx) diff --git a/metadata-ingestion/src/datahub/ingestion/source/snaplogic/snaplogic_config.py b/metadata-ingestion/src/datahub/ingestion/source/snaplogic/snaplogic_config.py new file mode 100644 index 00000000000000..df048f33ffd0bf --- /dev/null +++ b/metadata-ingestion/src/datahub/ingestion/source/snaplogic/snaplogic_config.py @@ -0,0 +1,33 @@ +from typing import Optional + +from pydantic import Field + +from datahub.ingestion.source.state.stale_entity_removal_handler import ( + StatefulIngestionConfigBase, + StatefulStaleMetadataRemovalConfig, +) +from datahub.ingestion.source.state.stateful_ingestion_base import ( + StatefulLineageConfigMixin, + StatefulUsageConfigMixin, +) + + +class SnaplogicConfig( + StatefulIngestionConfigBase, StatefulLineageConfigMixin, StatefulUsageConfigMixin +): + platform: str = "Snaplogic" + username: str = Field(description="Username") + password: str = Field(description="Password") + base_url: str = Field( + default="https://elastic.snaplogic.com", + description="Url to your Snaplogic instance: `https://elastic.snaplogic.com`, or similar. Used for making API calls to Snaplogic.", + ) + org_name: str = Field(description="Organization name from Snaplogic instance") + namespace_mapping: dict = Field( + default={}, description="Mapping of namespaces to platform instances" + ) + case_insensitive_namespaces: list = Field( + default=[], + description="List of namespaces that should be treated as case insensitive", + ) + stateful_ingestion: Optional[StatefulStaleMetadataRemovalConfig] = None diff --git a/metadata-ingestion/src/datahub/ingestion/source/snaplogic/snaplogic_lineage_extractor.py b/metadata-ingestion/src/datahub/ingestion/source/snaplogic/snaplogic_lineage_extractor.py new file mode 100644 index 00000000000000..7a21777a5c02ed --- /dev/null +++ b/metadata-ingestion/src/datahub/ingestion/source/snaplogic/snaplogic_lineage_extractor.py @@ -0,0 +1,104 @@ +from datetime import datetime +from typing import Iterable, Optional, Tuple + +import requests + +from datahub.ingestion.api.source import ( + SourceReport, +) +from datahub.ingestion.source.snaplogic.snaplogic_config import SnaplogicConfig +from datahub.ingestion.source.state.redundant_run_skip_handler import ( + RedundantLineageRunSkipHandler, +) + + +class SnaplogicLineageExtractor: + """ + A class to interact with the SnapLogic API. + """ + + def __init__( + self, + config: SnaplogicConfig, + redundant_run_skip_handler: Optional[RedundantLineageRunSkipHandler], + report: SourceReport, + ): + self.config = config + self.report = report + self.redundant_run_skip_handler = redundant_run_skip_handler + self.start_time, self.end_time = self._get_time_window() + + def get_lineages(self) -> Iterable[dict]: + """Generator function that yields lineage records one at a time as they are fetched.""" + page = 0 + has_more = True + records_processed = 0 + + try: + while has_more: + params = { + "format": "OPENLINEAGE", + "start_ts": str(int(self.start_time.timestamp() * 1000)), + "end_ts": str(int(self.end_time.timestamp() * 1000)), + "page": str(page), + } + + self.report.info( + message=f"Fetching lineage data - page: {page}, start_ts: {self.start_time}, end_ts: {self.end_time}", + title="Lineage Fetch", + ) + headers = {"User-Agent": "datahub-connector/1.0"} + response = requests.get( + url=f"{self.config.base_url}/api/1/rest/public/catalog/{self.config.org_name}/lineage", + params=params, + headers=headers, + auth=(self.config.username, self.config.password), + ) + response.raise_for_status() + + data = response.json() + content = data["content"] + + # Yield records one at a time + for record in content: + records_processed += 1 + yield record + + # Check if we need to fetch more pages + has_more = ( + len(content) >= 20 + ) # If we got full page size, there might be more + page += 1 + + self.report.info( + message=f"Completed fetching lineage data. Total records processed: {records_processed}", + title="Lineage Fetch Complete", + ) + + except Exception as e: + self.report.report_failure( + message="Error fetching lineage data", + exc=e, + title="Lineage Fetch Error", + ) + raise + + def _get_time_window(self) -> Tuple[datetime, datetime]: + if self.redundant_run_skip_handler: + return self.redundant_run_skip_handler.suggest_run_time_window( + self.config.start_time, self.config.end_time + ) + else: + return self.config.start_time, self.config.end_time + + def update_stats(self): + if self.redundant_run_skip_handler: + # Update the checkpoint state for this run. + self.redundant_run_skip_handler.update_state( + self.config.start_time, + self.config.end_time, + ) + + def report_status(self, step: str, status: bool) -> None: + if self.redundant_run_skip_handler: + self.redundant_run_skip_handler.report_current_run_status(step, status) diff --git a/metadata-ingestion/src/datahub/ingestion/source/snaplogic/snaplogic_parser.py b/metadata-ingestion/src/datahub/ingestion/source/snaplogic/snaplogic_parser.py new file mode 100644 index 00000000000000..6f0a053bc6dd04 --- /dev/null +++ b/metadata-ingestion/src/datahub/ingestion/source/snaplogic/snaplogic_parser.py @@ -0,0 +1,168 @@ +from dataclasses import dataclass +from typing import Dict, List, Optional + + +@dataclass +class Dataset: + name: str + display_name: str + fields: List[Dict] + platform: str = "snaplogic" + platform_instance: Optional[str] = None + type: Optional[str] = None # INPUT or OUTPUT + env: str = "PROD" + + +@dataclass +class Pipeline: + name: str + id: str + namespace: str + + +@dataclass +class Task: + name: str + id: str + namespace: str + + +@dataclass +class ColumnMapping: + input_dataset: Dataset + output_dataset: Dataset + input_field: str + output_field: str + + +class SnapLogicParser: + def __init__(self, case_insensitive_namespaces: list[str], namespace_mapping: dict): + self.case_insensitive_namespaces = case_insensitive_namespaces + self.namespace_mapping = namespace_mapping + self.platform_mapping = { + "sqlserver": "mssql", + } + + def _parse_platform(self, namespace: str) -> str: + type_part = namespace.split("://")[0] if "://" in namespace else namespace + + return self.platform_mapping.get(type_part.lower(), type_part.lower()) + + def extract_task_from_lineage(self, lineage: dict) -> Task: + job = lineage.get("job") + if not job: + raise ValueError("Job information is missing in the lineage data.") + name = job.get("name") + namespace = job.get("namespace") + + return Task( + id=name, + name=name.rsplit(":", 1)[0], + namespace=self._parse_platform(namespace), + ) + + def extract_pipeline_from_lineage(self, lineage: dict) -> Pipeline: + parent_run = lineage.get("run", {}).get("facets", {}).get("parent", {}) + job = parent_run.get("job", {}) + name = job.get("name") + namespace = job.get("namespace") + pipeline_snode_id = parent_run.get("_producer").split("#pipe_snode=")[1] + return Pipeline( + id=pipeline_snode_id, name=name, namespace=self._parse_platform(namespace) + ) + + def _get_case_sensitive_value(self, value: str, namespace: str) -> str: + """Transform value to lowercase if namespace is case-insensitive.""" + return value.lower() if namespace in self.case_insensitive_namespaces else value + + def _create_dataset_info( + self, + namespace: str, + name: str, + display_name: str, + type: str, + fields: Optional[List[Dict]] = None, + ) -> Dataset: + """Create a Dataset instance with proper case sensitivity.""" + return Dataset( + platform=self._parse_platform(namespace), + name=self._get_case_sensitive_value(name, namespace), + display_name=display_name or name, + fields=fields or [], + env="PROD", + platform_instance=self.namespace_mapping.get(namespace, None), + type=type, + ) + + def extract_columns_mapping_from_lineage( + self, lineage: dict + ) -> List[ColumnMapping]: + outputs = lineage.get("outputs", []) + lineages = [] + + for output in outputs: + output_namespace = output.get("namespace") + output_name = output.get("name", "") + column_lineage = ( + output.get("facets", {}).get("columnLineage", {}).get("fields", {}) + ) + + for field_name, field in column_lineage.items(): + output_field = self._get_case_sensitive_value( + field_name, output_namespace + ) + + for input_field in field.get("inputFields", []): + input_namespace = input_field.get("namespace") + input_name = input_field.get("name", "") + input_field_name = input_field.get("field", "") + + lineages.append( + ColumnMapping( + input_dataset=self._create_dataset_info( + input_namespace, input_name, input_name, "INPUT" + ), + output_dataset=self._create_dataset_info( + output_namespace, output_name, output_name, "OUTPUT" + ), + input_field=self._get_case_sensitive_value( + input_field_name, input_namespace + ), + output_field=output_field, + ) + ) + + return lineages + + def extract_datasets_from_lineage(self, lineage: dict) -> List[Dataset]: + inputs = lineage.get("inputs", {}) + outputs = lineage.get("outputs", {}) + + datasets = [] + for dataset, dataset_type in [ + *[(input_dataset, "INPUT") for input_dataset in inputs], + *[(output_dataset, "OUTPUT") for output_dataset in outputs], + ]: + namespace = dataset.get("namespace") + name = dataset.get("name", "") + fields = dataset.get("facets", {}).get("schema", {}).get("fields", []) + display_name = name + + # Transform names to lowercase if namespace is in case_insensitive_namespaces + if namespace in self.case_insensitive_namespaces: + name = name.lower() + fields = [ + {**field, "name": field.get("name", "").lower()} for field in fields + ] + + datasets.append( + self._create_dataset_info( + namespace=namespace, + name=name, + fields=fields, + display_name=display_name, + type=dataset_type, + ) + ) + + return datasets diff --git a/metadata-ingestion/tests/integration/snaplogic/__init__.py b/metadata-ingestion/tests/integration/snaplogic/__init__.py new file mode 100644 index 00000000000000..e69de29bb2d1d6 diff --git a/metadata-ingestion/tests/integration/snaplogic/snaplogic_base_golden.json b/metadata-ingestion/tests/integration/snaplogic/snaplogic_base_golden.json new file mode 100644 index 00000000000000..87c0197ac4d782 --- /dev/null +++ b/metadata-ingestion/tests/integration/snaplogic/snaplogic_base_golden.json @@ -0,0 +1,510 @@ +[ + { + "entityType": "dataFlow", + "entityUrn": "urn:li:dataFlow:(snaplogic,685013b9da1804dd3b4037e8,PROD)", + "changeType": "UPSERT", + "aspectName": "dataFlowInfo", + "aspect": { + "json": { + "customProperties": {}, + "externalUrl": "https://elastic.snaplogic.com/sl/designer.html?v=21818#pipe_snode=685013b9da1804dd3b4037e8", + "name": "Datahub Demo 3", + "description": "" + } + }, + "systemMetadata": { + "lastObserved": 1586847600000, + "runId": "test-snaplogic", + "lastRunId": "no-run-id-provided", + "pipelineName": "snaplogic_ingest" + } + }, + { + "entityType": "dataset", + "entityUrn": "urn:li:dataset:(urn:li:dataPlatform:mssql,snaplogic-test.tonyschema.accounts,PROD)", + "changeType": "UPSERT", + "aspectName": "datasetProperties", + "aspect": { + "json": { + "customProperties": {}, + "name": "snaplogic-test.tonyschema.accounts", + "qualifiedName": "snaplogic-test.tonyschema.accounts", + "tags": [] + } + }, + "systemMetadata": { + "lastObserved": 1586847600000, + "runId": "test-snaplogic", + "lastRunId": "no-run-id-provided", + "pipelineName": "snaplogic_ingest" + } + }, + { + "entityType": "dataset", + "entityUrn": "urn:li:dataset:(urn:li:dataPlatform:mssql,snaplogic-test.tonyschema.accounts,PROD)", + "changeType": "UPSERT", + "aspectName": "schemaMetadata", + "aspect": { + "json": { + "schemaName": "snaplogic-test.tonyschema.accounts", + "platform": "urn:li:dataPlatform:mssql", + "version": 0, + "created": { + "time": 0, + "actor": "urn:li:corpuser:unknown" + }, + "lastModified": { + "time": 0, + "actor": "urn:li:corpuser:unknown" + }, + "hash": "", + "platformSchema": { + "com.linkedin.schema.OtherSchema": { + "rawSchema": "" + } + }, + "fields": [ + { + "fieldPath": "Id", + "nullable": false, + "type": { + "type": { + "com.linkedin.schema.StringType": {} + } + }, + "nativeDataType": "VARCHAR", + "recursive": false, + "isPartOfKey": false + }, + { + "fieldPath": "IsDeleted", + "nullable": false, + "type": { + "type": { + "com.linkedin.schema.StringType": {} + } + }, + "nativeDataType": "VARCHAR", + "recursive": false, + "isPartOfKey": false + }, + { + "fieldPath": "MasterRecordId", + "nullable": false, + "type": { + "type": { + "com.linkedin.schema.StringType": {} + } + }, + "nativeDataType": "VARCHAR", + "recursive": false, + "isPartOfKey": false + }, + { + "fieldPath": "Name", + "nullable": false, + "type": { + "type": { + "com.linkedin.schema.StringType": {} + } + }, + "nativeDataType": "VARCHAR", + "recursive": false, + "isPartOfKey": false + }, + { + "fieldPath": "Type", + "nullable": false, + "type": { + "type": { + "com.linkedin.schema.StringType": {} + } + }, + "nativeDataType": "VARCHAR", + "recursive": false, + "isPartOfKey": false + }, + { + "fieldPath": "account_id", + "nullable": false, + "type": { + "type": { + "com.linkedin.schema.StringType": {} + } + }, + "nativeDataType": "VARCHAR", + "recursive": false, + "isPartOfKey": false + }, + { + "fieldPath": "contact_name", + "nullable": false, + "type": { + "type": { + "com.linkedin.schema.StringType": {} + } + }, + "nativeDataType": "VARCHAR", + "recursive": false, + "isPartOfKey": false + } + ] + } + }, + "systemMetadata": { + "lastObserved": 1586847600000, + "runId": "test-snaplogic", + "lastRunId": "no-run-id-provided", + "pipelineName": "snaplogic_ingest" + } + }, + { + "entityType": "dataset", + "entityUrn": "urn:li:dataset:(urn:li:dataPlatform:snaplogic,Virtual_DB.Virtual_Schema.Azure Synapse SQL - Select:2faf1,PROD)", + "changeType": "UPSERT", + "aspectName": "datasetProperties", + "aspect": { + "json": { + "customProperties": {}, + "name": "Virtual_DB.Virtual_Schema.Azure Synapse SQL - Select:2faf1", + "qualifiedName": "Virtual_DB.Virtual_Schema.Azure Synapse SQL - Select:2faf1", + "tags": [] + } + }, + "systemMetadata": { + "lastObserved": 1586847600000, + "runId": "test-snaplogic", + "lastRunId": "no-run-id-provided", + "pipelineName": "snaplogic_ingest" + } + }, + { + "entityType": "dataset", + "entityUrn": "urn:li:dataset:(urn:li:dataPlatform:snaplogic,Virtual_DB.Virtual_Schema.Azure Synapse SQL - Select:2faf1,PROD)", + "changeType": "UPSERT", + "aspectName": "schemaMetadata", + "aspect": { + "json": { + "schemaName": "Virtual_DB.Virtual_Schema.Azure Synapse SQL - Select:2faf1", + "platform": "urn:li:dataPlatform:snaplogic", + "version": 0, + "created": { + "time": 0, + "actor": "urn:li:corpuser:unknown" + }, + "lastModified": { + "time": 0, + "actor": "urn:li:corpuser:unknown" + }, + "hash": "", + "platformSchema": { + "com.linkedin.schema.OtherSchema": { + "rawSchema": "" + } + }, + "fields": [ + { + "fieldPath": "Id", + "nullable": false, + "type": { + "type": { + "com.linkedin.schema.StringType": {} + } + }, + "nativeDataType": "VARCHAR", + "recursive": false, + "isPartOfKey": false + }, + { + "fieldPath": "IsDeleted", + "nullable": false, + "type": { + "type": { + "com.linkedin.schema.StringType": {} + } + }, + "nativeDataType": "VARCHAR", + "recursive": false, + "isPartOfKey": false + }, + { + "fieldPath": "MasterRecordId", + "nullable": false, + "type": { + "type": { + "com.linkedin.schema.StringType": {} + } + }, + "nativeDataType": "VARCHAR", + "recursive": false, + "isPartOfKey": false + }, + { + "fieldPath": "Name", + "nullable": false, + "type": { + "type": { + "com.linkedin.schema.StringType": {} + } + }, + "nativeDataType": "VARCHAR", + "recursive": false, + "isPartOfKey": false + }, + { + "fieldPath": "Type", + "nullable": false, + "type": { + "type": { + "com.linkedin.schema.StringType": {} + } + }, + "nativeDataType": "VARCHAR", + "recursive": false, + "isPartOfKey": false + }, + { + "fieldPath": "account_id", + "nullable": false, + "type": { + "type": { + "com.linkedin.schema.StringType": {} + } + }, + "nativeDataType": "VARCHAR", + "recursive": false, + "isPartOfKey": false + }, + { + "fieldPath": "contact_name", + "nullable": false, + "type": { + "type": { + "com.linkedin.schema.StringType": {} + } + }, + "nativeDataType": "VARCHAR", + "recursive": false, + "isPartOfKey": false + } + ] + } + }, + "systemMetadata": { + "lastObserved": 1586847600000, + "runId": "test-snaplogic", + "lastRunId": "no-run-id-provided", + "pipelineName": "snaplogic_ingest" + } + }, + { + "entityType": "dataJob", + "entityUrn": "urn:li:dataJob:(urn:li:dataFlow:(snaplogic,685013b9da1804dd3b4037e8,PROD),Datahub Demo 3:Azure Synapse SQL - Select:2faf1220-44e8-4a52-a093-48f1e4d7d79a)", + "changeType": "UPSERT", + "aspectName": "dataJobInfo", + "aspect": { + "json": { + "customProperties": {}, + "externalUrl": "https://elastic.snaplogic.com/sl/designer.html?v=21818#pipe_snode=685013b9da1804dd3b4037e8", + "name": "Datahub Demo 3:Azure Synapse SQL - Select", + "description": "", + "type": { + "string": "SNAPLOGIC_SNAP" + } + } + }, + "systemMetadata": { + "lastObserved": 1586847600000, + "runId": "test-snaplogic", + "lastRunId": "no-run-id-provided", + "pipelineName": "snaplogic_ingest" + } + }, + { + "entityType": "dataJob", + "entityUrn": "urn:li:dataJob:(urn:li:dataFlow:(snaplogic,685013b9da1804dd3b4037e8,PROD),Datahub Demo 3:Azure Synapse SQL - Select:2faf1220-44e8-4a52-a093-48f1e4d7d79a)", + "changeType": "UPSERT", + "aspectName": "dataJobInputOutput", + "aspect": { + "json": { + "inputDatasets": [ + "urn:li:dataset:(urn:li:dataPlatform:mssql,snaplogic-test.tonyschema.accounts,PROD)" + ], + "outputDatasets": [ + "urn:li:dataset:(urn:li:dataPlatform:snaplogic,Virtual_DB.Virtual_Schema.Azure Synapse SQL - Select:2faf1,PROD)" + ], + "inputDatasetFields": [ + "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:mssql,snaplogic-test.tonyschema.accounts,PROD),Id)", + "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:mssql,snaplogic-test.tonyschema.accounts,PROD),IsDeleted)", + "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:mssql,snaplogic-test.tonyschema.accounts,PROD),MasterRecordId)", + "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:mssql,snaplogic-test.tonyschema.accounts,PROD),Name)", + "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:mssql,snaplogic-test.tonyschema.accounts,PROD),Type)", + "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:mssql,snaplogic-test.tonyschema.accounts,PROD),account_id)", + "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:mssql,snaplogic-test.tonyschema.accounts,PROD),contact_name)" + ], + "outputDatasetFields": [ + "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snaplogic,Virtual_DB.Virtual_Schema.Azure Synapse SQL - Select:2faf1,PROD),Id)", + "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snaplogic,Virtual_DB.Virtual_Schema.Azure Synapse SQL - Select:2faf1,PROD),IsDeleted)", + "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snaplogic,Virtual_DB.Virtual_Schema.Azure Synapse SQL - Select:2faf1,PROD),MasterRecordId)", + "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snaplogic,Virtual_DB.Virtual_Schema.Azure Synapse SQL - Select:2faf1,PROD),Name)", + "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snaplogic,Virtual_DB.Virtual_Schema.Azure Synapse SQL - Select:2faf1,PROD),Type)", + "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snaplogic,Virtual_DB.Virtual_Schema.Azure Synapse SQL - Select:2faf1,PROD),account_id)", + "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snaplogic,Virtual_DB.Virtual_Schema.Azure Synapse SQL - Select:2faf1,PROD),contact_name)" + ], + "fineGrainedLineages": [ + { + "upstreamType": "FIELD_SET", + "upstreams": [ + "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:mssql,snaplogic-test.tonyschema.accounts,PROD),IsDeleted)" + ], + "downstreamType": "FIELD_SET", + "downstreams": [ + "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snaplogic,Virtual_DB.Virtual_Schema.Azure Synapse SQL - Select:2faf1,PROD),IsDeleted)" + ], + "confidenceScore": 1.0 + }, + { + "upstreamType": "FIELD_SET", + "upstreams": [ + "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:mssql,snaplogic-test.tonyschema.accounts,PROD),Type)" + ], + "downstreamType": "FIELD_SET", + "downstreams": [ + "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snaplogic,Virtual_DB.Virtual_Schema.Azure Synapse SQL - Select:2faf1,PROD),Type)" + ], + "confidenceScore": 1.0 + }, + { + "upstreamType": "FIELD_SET", + "upstreams": [ + "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:mssql,snaplogic-test.tonyschema.accounts,PROD),contact_name)" + ], + "downstreamType": "FIELD_SET", + "downstreams": [ + "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snaplogic,Virtual_DB.Virtual_Schema.Azure Synapse SQL - Select:2faf1,PROD),contact_name)" + ], + "confidenceScore": 1.0 + }, + { + "upstreamType": "FIELD_SET", + "upstreams": [ + "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:mssql,snaplogic-test.tonyschema.accounts,PROD),account_id)" + ], + "downstreamType": "FIELD_SET", + "downstreams": [ + "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snaplogic,Virtual_DB.Virtual_Schema.Azure Synapse SQL - Select:2faf1,PROD),account_id)" + ], + "confidenceScore": 1.0 + }, + { + "upstreamType": "FIELD_SET", + "upstreams": [ + "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:mssql,snaplogic-test.tonyschema.accounts,PROD),Id)" + ], + "downstreamType": "FIELD_SET", + "downstreams": [ + "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snaplogic,Virtual_DB.Virtual_Schema.Azure Synapse SQL - Select:2faf1,PROD),Id)" + ], + "confidenceScore": 1.0 + }, + { + "upstreamType": "FIELD_SET", + "upstreams": [ + "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:mssql,snaplogic-test.tonyschema.accounts,PROD),MasterRecordId)" + ], + "downstreamType": "FIELD_SET", + "downstreams": [ + "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snaplogic,Virtual_DB.Virtual_Schema.Azure Synapse SQL - Select:2faf1,PROD),MasterRecordId)" + ], + "confidenceScore": 1.0 + }, + { + "upstreamType": "FIELD_SET", + "upstreams": [ + "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:mssql,snaplogic-test.tonyschema.accounts,PROD),Name)" + ], + "downstreamType": "FIELD_SET", + "downstreams": [ + "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snaplogic,Virtual_DB.Virtual_Schema.Azure Synapse SQL - Select:2faf1,PROD),Name)" + ], + "confidenceScore": 1.0 + } + ] + } + }, + "systemMetadata": { + "lastObserved": 1586847600000, + "runId": "test-snaplogic", + "lastRunId": "no-run-id-provided", + "pipelineName": "snaplogic_ingest" + } + }, + { + "entityType": "dataFlow", + "entityUrn": "urn:li:dataFlow:(snaplogic,685013b9da1804dd3b4037e8,PROD)", + "changeType": "UPSERT", + "aspectName": "status", + "aspect": { + "json": { + "removed": false + } + }, + "systemMetadata": { + "lastObserved": 1586847600000, + "runId": "test-snaplogic", + "lastRunId": "no-run-id-provided", + "pipelineName": "snaplogic_ingest" + } + }, + { + "entityType": "dataJob", + "entityUrn": "urn:li:dataJob:(urn:li:dataFlow:(snaplogic,685013b9da1804dd3b4037e8,PROD),Datahub Demo 3:Azure Synapse SQL - Select:2faf1220-44e8-4a52-a093-48f1e4d7d79a)", + "changeType": "UPSERT", + "aspectName": "status", + "aspect": { + "json": { + "removed": false + } + }, + "systemMetadata": { + "lastObserved": 1586847600000, + "runId": "test-snaplogic", + "lastRunId": "no-run-id-provided", + "pipelineName": "snaplogic_ingest" + } + }, + { + "entityType": "dataset", + "entityUrn": "urn:li:dataset:(urn:li:dataPlatform:mssql,snaplogic-test.tonyschema.accounts,PROD)", + "changeType": "UPSERT", + "aspectName": "status", + "aspect": { + "json": { + "removed": false + } + }, + "systemMetadata": { + "lastObserved": 1586847600000, + "runId": "test-snaplogic", + "lastRunId": "no-run-id-provided", + "pipelineName": "snaplogic_ingest" + } + }, + { + "entityType": "dataset", + "entityUrn": "urn:li:dataset:(urn:li:dataPlatform:snaplogic,Virtual_DB.Virtual_Schema.Azure Synapse SQL - Select:2faf1,PROD)", + "changeType": "UPSERT", + "aspectName": "status", + "aspect": { + "json": { + "removed": false + } + }, + "systemMetadata": { + "lastObserved": 1586847600000, + "runId": "test-snaplogic", + "lastRunId": "no-run-id-provided", + "pipelineName": "snaplogic_ingest" + } + } +] diff --git a/metadata-ingestion/tests/integration/snaplogic/snaplogic_base_recipe.yml b/metadata-ingestion/tests/integration/snaplogic/snaplogic_base_recipe.yml new file mode 100644 index 00000000000000..cbb3c3a30c105e --- /dev/null +++ b/metadata-ingestion/tests/integration/snaplogic/snaplogic_base_recipe.yml @@ -0,0 +1,16 @@ +pipeline_name: snaplogic_base_pipeline +source: + type: snaplogic + config: + username: example@snaplogic.com + password: dummy_password + base_url: https://elastic.snaplogic.com + org_name: TEST_ORG + stateful_ingestion: + enabled: True + remove_stale_metadata: False + +sink: + type: file + config: + filename: "/tmp/snaplogic_mces.json" diff --git a/metadata-ingestion/tests/integration/snaplogic/snaplogic_base_response.json b/metadata-ingestion/tests/integration/snaplogic/snaplogic_base_response.json new file mode 100644 index 00000000000000..03f06545b1c4c9 --- /dev/null +++ b/metadata-ingestion/tests/integration/snaplogic/snaplogic_base_response.json @@ -0,0 +1,213 @@ +{ + "size": 1, + "page": 0, + "content": [ + { + "eventTime": "2025-07-15T13:26:38.987Z", + "producer": "https://tahoe.elastic.snaplogicdev.com/sl/designer.html?#pipe_snode=685013b9da1804dd3b4037e8", + "schemaURL": "https://openlineage.io/spec/1-0-0/OpenLineage.json#/definitions/RunEvent", + "eventType": "COMPLETE", + "run": { + "runId": "99ad4ca0-c368-32a3-b573-e25f05d70d98", + "facets": { + "parent": { + "_producer": "https://tahoe.elastic.snaplogicdev.com/sl/designer.html?#pipe_snode=685013b9da1804dd3b4037e8", + "job": { + "namespace": "SnapLogic", + "name": "Datahub Demo 3" + }, + "_schemaURL": "https://openlineage.io/spec/facets/1-0-0/ParentRunFacet.json", + "run": { + "runId": "b519a833-0b1a-45b3-8dc6-ba35e103e015" + } + } + } + }, + "job": { + "namespace": "SnapLogic", + "name": "Datahub Demo 3:Azure Synapse SQL - Select:2faf1220-44e8-4a52-a093-48f1e4d7d79a" + }, + "inputs": [ + { + "namespace": "sqlserver://snaplogic-test.database.windows.net:1433", + "name": "snaplogic-test.tonyschema.accounts", + "facets": { + "schema": { + "_producer": "https://tahoe.elastic.snaplogicdev.com/sl/designer.html?#pipe_snode=685013b9da1804dd3b4037e8", + "fields": [ + { + "name": "Id", + "type": "VARCHAR", + "description": "" + }, + { + "name": "IsDeleted", + "type": "VARCHAR", + "description": "" + }, + { + "name": "MasterRecordId", + "type": "VARCHAR", + "description": "" + }, + { + "name": "Name", + "type": "VARCHAR", + "description": "" + }, + { + "name": "Type", + "type": "VARCHAR", + "description": "" + }, + { + "name": "account_id", + "type": "VARCHAR", + "description": "" + }, + { + "name": "contact_name", + "type": "VARCHAR", + "description": "" + } + ], + "_schemaURL": "https://openlineage.io/spec/facets/1-0-0/SchemaDatasetFacet.json" + } + } + } + ], + "outputs": [ + { + "namespace": "SnapLogic", + "name": "Virtual_DB.Virtual_Schema.Azure Synapse SQL - Select:2faf1", + "facets": { + "schema": { + "_producer": "https://tahoe.elastic.snaplogicdev.com/sl/designer .html?#pipe_snode=685013b9da1804dd3b4037e8", + "fields": [ + { + "name": "Id", + "type": "VARCHAR", + "description": "" + }, + { + "name": "IsDeleted", + "type": "VARCHAR", + "description": "" + }, + { + "name": "MasterRecordId", + "type": "VARCHAR", + "description": "" + }, + { + "name": "Name", + "type": "VARCHAR", + "description": "" + }, + { + "name": "Type", + "type": "VARCHAR", + "description": "" + }, + { + "name": "account_id", + "type": "VARCHAR", + "description": "" + }, + { + "name": "contact_name", + "type": "VARCHAR", + "description": "" + } + ], + "_schemaURL": "https://openlineage.io/spec/facets/1-0-0/SchemaDatasetFacet.json" + }, + "columnLineage": { + "_producer": "https://tahoe.elastic.snaplogicdev.com/sl/designer.html?#pipe_snode=685013b9da1804dd3b4037e8", + "fields": { + "IsDeleted": { + "inputFields": [ + { + "field": "IsDeleted", + "name": "snaplogic-test.tonyschema.accounts", + "namespace": "sqlserver://snaplogic-test.database.windows.net:1433" + } + ], + "transformationType": "IDENTITY", + "transformationDescription": "Mapping" + }, + "Type": { + "inputFields": [ + { + "field": "Type", + "name": "snaplogic-test.tonyschema.accounts", + "namespace": "sqlserver://snaplogic-test.database.windows.net:1433" + } + ], + "transformationType": "IDENTITY", + "transformationDescription": "Mapping" + }, + "contact_name": { + "inputFields": [ + { + "field": "contact_name", + "name": "snaplogic-test.tonyschema.accounts", + "namespace": "sqlserver://snaplogic-test.database.windows.net:1433" + } + ], + "transformationType": "IDENTITY", + "transformationDescription": "Mapping" + }, + "account_id": { + "inputFields": [ + { + "field": "account_id", + "name": "snaplogic-test.tonyschema.accounts", + "namespace": "sqlserver://snaplogic-test.database.windows.net:1433" + } + ], + "transformationType": "IDENTITY", + "transformationDescription": "Mapping" + }, + "Id": { + "inputFields": [ + { + "field": "Id", + "name": "snaplogic-test.tonyschema.accounts", + "namespace": "sqlserver://snaplogic-test.database.windows.net:1433" + } + ], + "transformationType": "IDENTITY", + "transformationDescription": "Mapping" + }, + "MasterRecordId": { + "inputFields": [ + { + "field": "MasterRecordId", + "name": "snaplogic-test.tonyschema.accounts", + "namespace": "sqlserver://snaplogic-test.database.windows.net:1433" + } + ], + "transformationType": "IDENTITY", + "transformationDescription": "Mapping" + }, + "Name": { + "inputFields": [ + { + "field": "Name", + "name": "snaplogic-test.tonyschema.accounts", + "namespace": "sqlserver://snaplogic-test.database.windows.net:1433" + } + ], + "transformationType": "IDENTITY", + "transformationDescription": "Mapping" + } + }, + "_schemaURL": "https://openlineage.io/spec/facets/1-0-1/ColumnLineageDatasetFacet.json" + } + } + } + ] + } + ] +} diff --git a/metadata-ingestion/tests/integration/snaplogic/test_snaplogic.py b/metadata-ingestion/tests/integration/snaplogic/test_snaplogic.py new file mode 100644 index 00000000000000..0a4e7f26029070 --- /dev/null +++ b/metadata-ingestion/tests/integration/snaplogic/test_snaplogic.py @@ -0,0 +1,106 @@ +import json +import pathlib +from typing import Any +from unittest.mock import patch + +from freezegun import freeze_time + +from datahub.ingestion.run.pipeline import Pipeline +from datahub.testing import mce_helpers + +FROZEN_TIME = "2020-04-14 07:00:00" +ORG_NAME = "TEST_ORG" + + +def default_recipe(tmp_path, output_file_name="snaplogic_mces_default_config.json"): + return { + "run_id": "test-snaplogic", + "pipeline_name": "snaplogic_ingest", + "source": { + "type": "snaplogic", + "config": { + "username": "example@snaplogic.com", + "password": "dummy_password", + "base_url": "https://elastic.snaplogic.com", + "org_name": ORG_NAME, + "stateful_ingestion": { + "enabled": False, + "remove_stale_metadata": False, + }, + }, + }, + "sink": { + "type": "file", + "config": { + "filename": f"{tmp_path}/{output_file_name}", + }, + }, + } + + +def register_mock_api(pytestconfig: Any, request_mock: Any) -> None: + test_resources_dir: pathlib.Path = ( + pytestconfig.rootpath / "tests/integration/snaplogic" + ) + + # Load the mock response from snaplogic_base_response.json + with open(test_resources_dir / "snaplogic_base_response.json", "r") as f: + snaplogic_response = json.load(f) + + api_vs_response = { + f"https://elastic.snaplogic.com/api/1/rest/public/catalog/{ORG_NAME}/lineage": { + "method": "GET", + "status_code": 200, + "json": snaplogic_response, + }, + } + + for url in api_vs_response: + request_mock.register_uri( + api_vs_response[url]["method"], + url, + json=api_vs_response[url]["json"], + status_code=api_vs_response[url]["status_code"], + ) + + +def run_ingest( + pytestconfig, + mock_datahub_graph, + recipe, +): + with ( + patch( + "datahub.ingestion.source.state_provider.datahub_ingestion_checkpointing_provider.DataHubGraph", + mock_datahub_graph, + ) as mock_checkpoint, + ): + mock_checkpoint.return_value = mock_datahub_graph + + # Run an azure usage ingestion run. + pipeline = Pipeline.create(recipe) + pipeline.run() + pipeline.raise_from_status() + return pipeline + + +@freeze_time(FROZEN_TIME) +def test_snaplogic_source_default_configs( + pytestconfig, mock_datahub_graph, tmp_path, requests_mock +): + test_resources_dir: pathlib.Path = ( + pytestconfig.rootpath / "tests/integration/snaplogic" + ) + register_mock_api(pytestconfig, requests_mock) + + run_ingest( + pytestconfig=pytestconfig, + mock_datahub_graph=mock_datahub_graph, + recipe=default_recipe(tmp_path), + ) + + mce_helpers.check_golden_file( + pytestconfig, + output_path=tmp_path / "snaplogic_mces_default_config.json", + golden_path=test_resources_dir / "snaplogic_base_golden.json", + )