I am looking to ingest data from Salesforce, I noticed that they have an API to describe all objects within Salesforce, and the response looks like this

{
  "size": 112,
  "totalSize": 112,
  "done": true,
  "queryLocator": null,
  "entityTypeName": "FieldDefinition",
  "records": [
    {
      "attributes": {
        "type": "FieldDefinition",
        "url": "/services/data/v53.0/tooling/sobjects/FieldDefinition/MessagingSession.Id"
      },
      "DataType": "Lookup()",
      "Description": null
    },
    {
      "attributes": {
        "type": "FieldDefinition",
        "url": "/services/data/v53.0/tooling/sobjects/FieldDefinition/MessagingSession.Owner"
      },
      "DataType": "Lookup(User,Group)",
      "Description": null
    },
    {
      "attributes": {
        "type": "FieldDefinition",
        "url": "/services/data/v53.0/tooling/sobjects/FieldDefinition/MessagingSession.IsDeleted"
      },
      "DataType": "Checkbox",
      "Description": null
    },
    {
      "attributes": {
        "type": "FieldDefinition",
        "url": "/services/data/v53.0/tooling/sobjects/FieldDefinition/MessagingSession.Name"
      },
      "DataType": "Auto Number",
      "Description": null
    }
  ]
}

As you can see the DataType is not similar to what we have in other languages. Would datahub be able to handle this if I manually ingest this by putting this in a ingestable file?

Hey there @red-pizza-28006, Datahub schemas have a String field

nativeDataType

that you may be looking for. This lets you provide an unstructured string to represent the data type if datahub's type enum isn't useful

@green-football-43791 Do you have any example of how I can read this API response and convert it into a Datahub ingestable file?

here's how we convert sql sources to datahub-friendly formats:

is there a sample file that I can see which i can transform this json to so Datahub can consume it?

yes, the file I shared with you contains that transformation. You can see how it constructs a Dataset that datahub understands here: https://github.com/linkedin/datahub/blob/master/metadata-ingestion/src/datahub/ingestion/source/sql/sql_common.py#L438

@green-football-43791 to emit this dataset, do I need to setup the entire Source class, or can I just emit it as it is inside my airflow DAG? I created some dummy code like this

datasetUrn = f"urn:li:dataset:(urn:li:dataPlatform:{platform},{dataset_name},{env})"
            dataset_snapshot = DatasetSnapshot(
                urn=datasetUrn,
                aspects=[],
            )
            mce = MetadataChangeEvent(proposedSnapshot=dataset_snapshot)
            wu = SfWorkUnit(id=dataset_name, mce=mce)
            self.report.report_workunit(wu)

But I dont see the dataset in datahub, probably need to use the sink somehow?

Ah I see- for emitting custom MCEs from an airflow dag, refer to the DatahubEmitterOperator that we supply: https://github.com/linkedin/datahub/blob/master/metadata-ingestion/src/datahub_provider/example_dags/lineage_emission_dag.py#L54

thanks, I tried some dummy code to see how much I need to code, I was getting this exception from my DAG, any idea what am I missing?

Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1138, in _run_raw_task
    self._prepare_and_execute_task_with_callbacks(context, task)
  File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1311, in _prepare_and_execute_task_with_callbacks
    result = self._execute_task(context, task_copy)
  File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1336, in _execute_task
    result = task_copy.execute(context=context)
  File "/usr/local/airflow/.local/lib/python3.7/site-packages/datahub_provider/operators/datahub.py", line 63, in execute
    self.generic_hook.get_underlying_hook().emit_mces(self.mces)
  File "/usr/local/airflow/.local/lib/python3.7/site-packages/datahub_provider/hooks/datahub.py", line 80, in emit_mces
    emitter.emit_mce(mce)
  File "/usr/local/airflow/.local/lib/python3.7/site-packages/datahub/emitter/rest_emitter.py", line 128, in emit_mce
    raw_mce_obj = mce.proposedSnapshot.to_obj()
  File "/usr/local/airflow/.local/lib/python3.7/site-packages/avrogen/dict_wrapper.py", line 45, in to_obj
    return conv.to_json_object(self, self.RECORD_SCHEMA)
  File "/usr/local/airflow/.local/lib/python3.7/site-packages/avrogen/avrojson.py", line 110, in to_json_object
    raise io.AvroTypeException(writers_schema, data_obj)
AttributeError: module '<http://avro.io|avro.io>' has no attribute 'AvroTypeException'

My MCE looks like this

local-runner_1  | MetadataChangeEventClass({'auditHeader': None, 'proposedSnapshot': DatasetSnapshotClass({'urn': 'urn:li:dataset:(urn:li:dataPlatform:salesforce,MessagingSession,DEV)', 'aspects': [SchemaMetadataClass({'schemaName': 'my_test', 'platform': 'urn:li:dataPlatform:salesforce', 'version': 0, 'created': AuditStampClass({'time': 0, 'actor': 'urn:li:corpuser:unknown', 'impersonator': None}), 'lastModified': AuditStampClass({'time': 0, 'actor': 'urn:li:corpuser:unknown', 'impersonator': None}), 'deleted': None, 'dataset': None, 'cluster': None, 'hash': '', 'platformSchema': MySqlDDLClass({'tableSchema': 'test'}), 'fields': SchemaFieldClass({'fieldPath': 'id', 'jsonPath': None, 'nullable': False, 'description': None, 'type': <class 'datahub.metadata.schema_classes.StringTypeClass'>, 'nativeDataType': 'LookUp()', 'recursive': False, 'globalTags': None, 'glossaryTerms': None, 'isPartOfKey': False}), 'primaryKeys': None, 'foreignKeysSpecs': None, 'foreignKeys': None})]}), 'proposedDelta': None, 'systemMetadata': None})

and here is my dummy code

data = Variable.get('sf_api_response', deserialize_json=True)
    dataset_name = 'MessagingSession'
    platform = "salesforce"
    env = "DEV"
    dataset_urn = f"urn:li:dataset:(urn:li:dataPlatform:{platform},{dataset_name},{env})"
    fields = SchemaField(
        fieldPath="id",
        type=StringTypeClass,
        nativeDataType="LookUp()",
    )
    schema_metadata = SchemaMetadata(
        schemaName="my_test",
        platform=f"urn:li:dataPlatform:{platform}",
        version=0,
        fields=fields,
        platformSchema=MySqlDDLClass("test"),
        hash=""
    )
    mce = MetadataChangeEventClass(
        proposedSnapshot=DatasetSnapshotClass(
            urn=dataset_urn,
            aspects=[schema_metadata],
        )
    )
    print(mce.items())

    test_task = DatahubEmitterOperator(
        task_id="emit_lineage",
        datahub_conn_id="datahub_rest_default",
        mces=[mce],
    )

    test_task

Which line is it particularly that is causing you the issues?

type=StringTypeClass

are you sure that

record['QualifiedApiName']

and

record['DataType']

are non-null?

yes it was one of the imported classes from schema_classes.py

oh, sure let me give it a try and see, thanks for the pointer 😬