feat(integration/airbyte): Airbyte source ingestion integration #11

shubhamjagtap639 · 2023-07-25T13:54:46Z

Checklist

The PR conforms to DataHub's Contributing Guideline (particularly Commit Message Format)
Links to related issues (if applicable)
Tests for the changes have been added/updated (if applicable)
Docs related to the changes have been added/updated (if applicable). If a new feature has been added a Usage Guide has been added for the same.
For any breaking change/potential downtime/deprecation/big changes an entry has been made in Updating DataHub

siddiquebagwan

Initial review comments

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

siddiquebagwan · 2023-07-27T09:19:23Z

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

+@config_class(AirbyteConfig)
+@support_status(SupportStatus.CERTIFIED)
+@capability(SourceCapability.PLATFORM_INSTANCE, "Enabled by default")
+class AirbyteSource(Source):


Enable stateful ingsetion

siddiquebagwan · 2023-07-27T09:21:12Z

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

+class AirbyteSource(Source):
+    """
+    This plugin extracts airbyte workspace, connections, sources, destinations and jobs.
+    This plugin is in beta and has only been tested on PostgreSQL.


Could you please add a description related to when to provide api_key vs username/password?

Added that in source config description

siddiquebagwan · 2023-07-31T07:44:20Z

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

+
+@platform_name("Airbyte")
+@config_class(AirbyteSourceConfig)
+@support_status(SupportStatus.CERTIFIED)


Change it to SupportStatus.INCUBATING

siddiquebagwan · 2023-07-31T08:02:06Z

metadata-ingestion/src/datahub/ingestion/source/airbyte/config.py

+class AirbyteSourceConfig(StatefulIngestionConfigBase, DatasetSourceConfigMixin):
+    cloud_deploy: bool = pydantic.Field(
+        default=False,
+        description="Whether to fetch metadata from Airbyte Cloud or Airbyte OSS. For Airbyte Cloud provide api_key and for Airbyte OSS provide username/password",


Enabled to fetch metadata from Airbyte Cloud. If it is enabled then provide api_key in the recipe. username & password is required for Airbyte OSS only.

siddiquebagwan · 2023-07-31T08:02:36Z

metadata-ingestion/src/datahub/ingestion/source/airbyte/config.py

+
+
+class AirbyteSourceConfig(StatefulIngestionConfigBase, DatasetSourceConfigMixin):
+    cloud_deploy: bool = pydantic.Field(


Add a method to validate if cloud_deploye is true then api_key should be present, Check powerbi config class

siddiquebagwan · 2023-07-31T08:15:01Z

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

+        config = AirbyteSourceConfig.parse_obj(config_dict)
+        return cls(config, ctx)
+
+    def get_workspace_workunit(


why this pass method ?

Removed pass and added code in that method

siddiquebagwan · 2023-07-31T10:15:00Z

metadata-ingestion/src/datahub/ingestion/source/airbyte/rest_api_wrapper/data_resolver.py

+
+
+class OssAPIResolver(DataResolverBase):
+    BASE_URL = "http://localhost:8000/api/v1"


take this from config and keep default to this value

siddiquebagwan · 2023-08-01T13:10:02Z

metadata-ingestion/src/datahub/ingestion/api/source_helpers.py

+            # If any entity does not support aspect 'status' then skip that entity from adding status aspect.
+            # Example like dataProcessInstance doesn't suppport status aspect.
+            # If not skipped gives error: java.lang.RuntimeException: Unknown aspect status for entity dataProcessInstance
+            skip_urns.add(urn)


We can say continue instead of collecting urns in skip_urns array

At source_helpers.py:100 we are yielding all workunits which needs to ingest, hence we cannot have continue statement at source_helpers.py:98.
We can have CONTINUE logic next to source:helpers.py:102 but there we need to create logic to extract entity type from string urn.
I think current changes are small and simple so lets keep it otherwise let me know we can modify it.

lets discuss this in call

siddiquebagwan · 2023-08-01T13:15:47Z

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

+                self.config.connector_platform_details[connector_type]
+            )
+        else:
+            connector_platform_detail = PlatformDetail()


Why this else and empty PlatformDetail?

Empty PlatformDetails will take default platform instance and env.
I will remove the else part and initialization of connector_platform_details will be done prior to if condition.

siddiquebagwan · 2023-08-01T13:18:35Z

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

+
+        return DatasetUrn.create_from_ids(
+            platform_id=supported_data_platform[connector_type],
+            table_name=connector.name,


Is it correct as per your concept mapping

Yes correct. We decided to map source/destination as a input/output dataset of datajob entity.
I forgot to add it inside concept mapping table but already mentioned in Approch paragraph.

siddiquebagwan · 2023-08-01T13:23:24Z

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

+            logger.info(f"Processing workspace id: {workspace.workspace_id}")
+            yield from self.get_workspace_workunit(workspace)
+
+    def get_report(self) -> SourceReport:


Check powerbi, add one test connection method

I already added the test connection methods for both cloud and oss in airbyte/rest_api_wrapper/data_resolver.py.
And using it at airbyte_api.py:33.

siddiquebagwan · 2023-08-03T09:42:17Z

metadata-ingestion/src/datahub/ingestion/api/source_helpers.py

+            # If any entity does not support aspect 'status' then skip that entity from adding status aspect.
+            # Example like dataProcessInstance doesn't suppport status aspect.
+            # If not skipped gives error: java.lang.RuntimeException: Unknown aspect status for entity dataProcessInstance
+            skip_urns.add(urn)


lets discuss this in call

siddiquebagwan · 2023-08-03T09:53:20Z

metadata-ingestion/src/datahub/ingestion/source/airbyte/rest_api_wrapper/data_resolver.py

+        )
+        if response.status_code == 401:
+            raise ConfigurationError(
+                "Please check if provided api key is correct or not."


Configured credentials don't have required permissions.

siddiquebagwan · 2023-08-03T09:54:08Z

metadata-ingestion/src/datahub/ingestion/source/airbyte/rest_api_wrapper/data_resolver.py

+        if response.status_code == 401:
+            raise ConfigurationError(
+                "Please check if provided api key is correct or not."
+            )


@shubhamjagtap639 Please document what minimum permission user or API key needs to fetch metadata

There is no any permission thing present at API level

siddiquebagwan-gslab · 2023-08-07T10:29:28Z

metadata-ingestion/docs/sources/airbyte/airbyte_pre.md

+## Configuration Notes
+1. Airbyte source is available for both Airbyte Cloud and Airbyte Open Source (OSS) users.
+2. For Airbyte Cloud user need to provide api_key in recipe to ingest metadata and for Airbyte OSS username and password.
+3. Refer Walkthrough demo [here](https://www.loom.com/share/7997a7c67cd642cc8d1c72ef0dfcc4bc) to create a api_key from [Developer Portal](https://portal.airbyte.com/) in case you are using Airbyte Cloud.


For Airbyte Cloud refer demo here to create a api_key.

siddiquebagwan-gslab · 2023-08-07T10:32:12Z

metadata-ingestion/docs/sources/airbyte/airbyte_pre.md

@@ -0,0 +1,16 @@
+## Configuration Notes
+1. Airbyte source is available for both Airbyte Cloud and Airbyte Open Source (OSS) users.
+2. For Airbyte Cloud user need to provide api_key in recipe to ingest metadata and for Airbyte OSS username and password.


Airbyte Cloud users need to provide api_key in the recipe for authentication and Airbyte OSS users need to provide username and password.

siddiquebagwan-gslab · 2023-08-07T10:33:21Z

metadata-ingestion/docs/sources/airbyte/airbyte_pre.md

+|--------------------------|-----------------------|
+| `Workspace`              | `DataFlow`       	   |
+| `Connection`             | `DataJob`             |
+| `Sourc`                  | `Dataset`             |


siddiquebagwan-gslab · 2023-08-07T10:36:54Z

metadata-ingestion/setup.py

@@ -377,6 +377,7 @@ def get_long_description():
    "powerbi-report-server": powerbi_report_server,
    "vertica": sql_common | {"vertica-sqlalchemy-dialect[vertica-python]==0.0.1"},
    "unity-catalog": databricks | sqllineage_lib,
+    "airbyte": {"requests"},


requests is a common library, check if it might be already there in this setup.py

No, If you see dbt-cloud and plusar source they are also defined in same way.

siddiquebagwan-gslab · 2023-08-07T10:47:26Z

metadata-ingestion/src/datahub/ingestion/source/airbyte/rest_api_wrapper/data_resolver.py

+
+    def _get_jobs_from_response(self, response: List[Dict]) -> List[Job]:
+        jobs: List[Job] = [
+            Job(


what is the purpose of these if else? lets discuss this in call

Aibyte cloud and OSS response contains some same metadata with different key.
Instead of defining seperate function in both CloudApiResolver and OssApiResolver I did this and define one function in abstraction class.

define two separate method please , eventually they will get call as per instance

siddiquebagwan-gslab · 2023-08-08T06:57:10Z

metadata-ingestion/docs/sources/airbyte/airbyte_pre.md

+| `Destination`            | `Dataset`             |
+| `Connection Job History` | `DataProcessInstance` |
+
+Source and destination gets mapped with Dataset as an Input and Output of Connection.


Source and destination are mapped to Dataset as an Input and Output of Connection.

siddiquebagwan-gslab · 2023-08-08T06:57:30Z

metadata-ingestion/setup.py

@@ -377,6 +377,7 @@ def get_long_description():
    "powerbi-report-server": powerbi_report_server,
    "vertica": sql_common | {"vertica-sqlalchemy-dialect[vertica-python]==0.0.1"},
    "unity-catalog": databricks | sqllineage_lib,
+    "airbyte": {"requests"},


siddiquebagwan-gslab · 2023-08-08T06:59:48Z

metadata-ingestion/src/datahub/ingestion/source/airbyte/rest_api_wrapper/data_resolver.py

+
+    def _get_jobs_from_response(self, response: List[Dict]) -> List[Job]:
+        jobs: List[Job] = [
+            Job(


define two separate method please , eventually they will get call as per instance

siddiquebagwan-gslab · 2023-08-08T07:02:14Z

metadata-ingestion/tests/integration/airbyte/test_airbyte.py

+from tests.test_helpers import mce_helpers
+
+
+def enable_logging():


remove this no need.

…into airbyte-source-integration

Airbyte initial Code added

88f049d

shubhamjagtap639 requested a review from siddiquebagwan July 25, 2023 13:54

siddiquebagwan requested changes Jul 27, 2023

View reviewed changes

Airbyte API and Stateful ingestion code added

a1a778b

siddiquebagwan requested changes Jul 31, 2023

View reviewed changes

Code to generate entity and get workunit added

59b572e

siddiquebagwan requested changes Aug 1, 2023

View reviewed changes

Test cases added for airbyte connector

9d39abf

siddiquebagwan requested changes Aug 3, 2023

View reviewed changes

Server to platform instance config added and test case modified

439fbd4

siddiquebagwan-gslab suggested changes Aug 7, 2023

View reviewed changes

Changes as per review comments

c9c071e

siddiquebagwan-gslab suggested changes Aug 8, 2023

View reviewed changes

shubhamjagtap639 and others added 4 commits August 8, 2023 14:40

Code changes as per PR review comments

b04f734

Merge branch 'master' of https://github.com/shubhamjagtap639/datahub …

fe82b2e

…into airbyte-source-integration

Resolve Merge Conflicts

2d39d35

Merge branch 'master' into airbyte-source-integration

4983ef3



		class AirbyteSourceConfig(StatefulIngestionConfigBase, DatasetSourceConfigMixin):
		cloud_deploy: bool = pydantic.Field(



		class OssAPIResolver(DataResolverBase):
		BASE_URL = "http://localhost:8000/api/v1"

		from tests.test_helpers import mce_helpers


		def enable_logging():

feat(integration/airbyte): Airbyte source ingestion integration #11

Are you sure you want to change the base?

feat(integration/airbyte): Airbyte source ingestion integration #11

Conversation

shubhamjagtap639 commented Jul 25, 2023

Checklist

siddiquebagwan left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment