Pushing Data Quality Check Results to DataHub using Python SDK or REST API through examples

ingestion
1

Original Slack Thread

Hello, team! Anyone has an example for pushing data quality check results (assertions) to DataHub using the Python SDK or the REST API?

2

Hi <@U04QE11EXDE>,

yes we do so via the Python SDK. What aspects are you interested about?
I can share some examples but this requires me to prepare the code, so I’d like to make sure to provide you with something useful for you :slightly_smiling_face:

3

<@U049WUH4155> I’m not using Great Expectations, I built a data quality system integrated to my Data Lake and after running the quality checks, I want to publish the results to DataHub, considering that I already have the quality check results ready and I just need to build the assertion objects and publish them to Datahub

4

So the only part I would like some example is on the correct way of building and publishing the results to Datahub

5

Interesting, that’s exactly what we did as well
I’m happy to provide you with some examples

6

In general you need to send. three kind of objects to the API:
• Assertion Info
• Assertion Platform MCP
• Assertion Result
To send the assertion info the following code is required:

            entityUrn=assertion.urn,
            aspect=<http://assertion.info|assertion.info>,
        )
emitter.emit_mcp(mcp=assertion_info_mcp)```
The `info` passed as aspect here is an instance of `<http://datahub.metadata.com|datahub.metadata.com>.linkedin.pegasus2avro.assertion.AssertionInfo`
How this should be created highly depends on the type of assertions you want to generate, so feel free to ask for more details.

Secondly, you can publish the platform assertion mapping as follows:
```assertion_platform_mcp = MetadataChangeProposalWrapper(
            entityUrn=assertion.urn,
            aspect=DataPlatformInstanceClass(platform="urn:li:dataPlatform:&lt;any-name-for-your-data-quality-platform&gt;"),
        )
emitter.emit_mcp(mcp=assertion_platform_mcp)```
And finally the result of the assertion:
``` assertion_result_mcp = MetadataChangeProposalWrapper(
            entityUrn=assertion.urn,
            aspect=assertion.result,
        )
emitter.emit_mcp(mcp=assertion_result_mcp)```
Here `result` is an instance of `<http://datahub.metadata.com|datahub.metadata.com>.linkedin.pegasus2avro.assertion.AssertionRunEvent` and can look like the following:
``` AssertionRunEvent(
            timestampMillis=int(round(time.time() * 1000)),
            assertionUrn=assertion_urn,
            asserteeUrn=dataset_urn,
            runId=datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ"),
            result=AssertionResult(
                type=AssertionResultType.SUCCESS if assertion_result.is_success else AssertionResultType.FAILURE,
                nativeResults={"observed_value": assertion_result.observed_value},
            ),
            status=AssertionRunStatus.COMPLETE,
        )```
I'm happy to go more into detail if required :slightly_smiling_face: