in the bootstrap sample mce list you can see an example of that here: https://github.com/linkedin/datahub/blob/master/metadata-ingestion/examples/mce_files/bootstrap_mce.json#L732

In that MCE, we are tagging the dataset as having a specific property

here is an example of us tagging a schema field as being `Legacy`:

and here is the MCE emitted to provide a definition for the Legacy tag:

Hope these examples help.

We don't yet have a reference implementation of a

TagApplier

transformer ... but @gray-shoe-75895 was planning to write one.

in the end, the MCE should look like one of the examples @green-football-43791 pointed out.

@gray-shoe-75895 @mammoth-bear-12532 @green-football-43791 Something like this?

from datahub.ingestion.api.transform import Transformer

from datahub.metadata import MetadataChangeEventClass, DatasetSnapshotClass, \

SchemaMetadataClass, SchemaFieldClass, SchemaFieldDataTypeClass, TagAssociationClass, GlobalTagsClass

from logging import getLogger

LOGGER = getLogger(__file__)

TABLE_TAGS = {

"urn:li:dataset:(urn:li:dataPlatform:postgresql,myapp.information_schema.sql_parts,PROD)": {"comments" : { "tags": ["a_tag", "another_tag"],

"description": "Description for comments" }}

}

class TagApplier(Transformer):

def __init__(self, config):

self.config = config

def transform(self, record_envelopes):

for envelope in record_envelopes:

if isinstance(envelope.record, MetadataChangeEventClass):

if isinstance(envelope.record.proposedSnapshot, DatasetSnapshotClass):

urn = envelope.record.proposedSnapshot.urn

if urn in TABLE_TAGS:

for aspect in envelope.record.proposedSnapshot.aspects:

if isinstance(aspect, SchemaMetadataClass):

for field in aspect.fields:

if isinstance(field, SchemaFieldClass):

if field.fieldPath in TABLE_TAGS[urn]:

if 'description' in TABLE_TAGS[urn][field.fieldPath]:

desc = TABLE_TAGS[urn][field.fieldPath]['description']

<http://LOGGER.info|LOGGER.info>("Setting table %s field %s description: %s", urn, field.fieldPath, desc)

field.description = TABLE_TAGS[urn][field.fieldPath]['description']

if 'tags' in TABLE_TAGS[urn][field.fieldPath]:

tags = TABLE_TAGS[urn][field.fieldPath]['tags']

<http://LOGGER.info|LOGGER.info>("Setting table %s field %s tags: %s", urn, field.fieldPath, tags)

field.globalTags = GlobalTagsClass(tags=[TagAssociationClass(tag=t) for t in tags])

print(envelope.record, type(envelope.record))

print(envelope.record.proposedSnapshot, type(envelope.record.proposedSnapshot))

print(envelope.record.proposedSnapshot.urn, type(envelope.record.proposedSnapshot.urn))

yield envelope

@classmethod

def create(cls, config_dict, ctx):

return TagApplier(config_dict)