Fixes #5448: Implement initial Iceberg Connector using PyIceberg

ingestion/setup.py

@@ -20,16 +20,19 @@
 # Add here versions required for multiple plugins
 VERSIONS = {
     "airflow": "apache-airflow==2.7.3",
+    "adlfs": "adlfs==2022.11.2",  # Python 3.7 does only support up to 2022.2.0
     "avro": "avro~=1.11",
     "boto3": "boto3>=1.20,<2.0",  # No need to add botocore separately. It's a dep from boto3
     "geoalchemy2": "GeoAlchemy2~=0.12",
     "google-cloud-storage": "google-cloud-storage==1.43.0",
+    "gcsfs": "gcsfs==2022.11.0",
     "great-expectations": "great-expectations~=0.18.0",
     "grpc-tools": "grpcio-tools>=1.47.2",
     "msal": "msal~=1.2",
     "neo4j": "neo4j~=5.3.0",
     "pandas": "pandas<=2,<3",
     "pyarrow": "pyarrow~=14.0",
+    "pydantic": "pydantic==1.10.13",
     "pydomo": "pydomo~=0.3",
     "pymysql": "pymysql>=1.0.2",
     "pyodbc": "pyodbc>=4.0.35,<5",
@@ -106,7 +109,7 @@
     "jsonschema",
     "memory-profiler",
     "mypy_extensions>=0.4.3",
-    "pydantic~=1.10",
+    VERSIONS["pydantic"],
     VERSIONS["pymysql"],
     "python-dateutil>=2.8.1",
     "python-jose~=3.3",
@@ -161,12 +164,12 @@
     "datalake-azure": {
         VERSIONS["azure-storage-blob"],
         VERSIONS["azure-identity"],
-        "adlfs>=2022.2.0",  # Python 3.7 does only support up to 2022.2.0
+        VERSIONS["adlfs"],  # Python 3.7 does only support up to 2022.2.0
         *COMMONS["datalake"],
     },
     "datalake-gcs": {
         VERSIONS["google-cloud-storage"],
-        "gcsfs==2022.11.0",
+        VERSIONS["gcsfs"],
         *COMMONS["datalake"],
     },
     "datalake-s3": {
@@ -197,6 +200,13 @@
         "thrift-sasl~=0.4",
         "impyla~=0.18.0",
     },
+    "iceberg": {
+        "pyiceberg",
+        # Forcing the version of a few packages so it plays nicely with other requirements.
+        VERSIONS["pydantic"],
+        VERSIONS["adlfs"],
+        VERSIONS["gcsfs"],
+    },
     "impala": {
         "presto-types-parser>=0.0.2",
         "impyla[kerberos]~=0.18.0",

ingestion/src/metadata/ingestion/source/database/iceberg/catalog/__init__.py

@@ -0,0 +1,49 @@
+#  Copyright 2021 Collate
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#  http://www.apache.org/licenses/LICENSE-2.0
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+"""
+Iceberg Catalog Factory.
+"""
+from typing import Dict, Type
+from pyiceberg.catalog import Catalog
+
+from metadata.generated.schema.entity.services.connections.database.icebergConnection import (
+    Catalog as IcebergCatalog,
+    HiveCatalogConnection,
+    RestCatalogConnection,
+    GlueCatalogConnection,
+    DynamoDbCatalogConnection,
+)
+from metadata.ingestion.source.database.iceberg.catalog.base import IcebergCatalogBase
+from metadata.ingestion.source.database.iceberg.catalog.dynamodb import IcebergDynamoDbCatalog
+from metadata.ingestion.source.database.iceberg.catalog.glue import IcebergGlueCatalog
+from metadata.ingestion.source.database.iceberg.catalog.hive import IcebergHiveCatalog
+from metadata.ingestion.source.database.iceberg.catalog.rest import IcebergRestCatalog
+
+
+
+class IcebergCatalogFactory:
+    catalog_type_map: Dict[str, Type[IcebergCatalogBase]] = {
+        RestCatalogConnection.__name__: IcebergRestCatalog,
+        HiveCatalogConnection.__name__: IcebergHiveCatalog,
+        GlueCatalogConnection.__name__: IcebergGlueCatalog,
+        DynamoDbCatalogConnection.__name__: IcebergDynamoDbCatalog,
+    }
+
+    @classmethod
+    def from_connection(cls, catalog: IcebergCatalog) -> Catalog:
+        """ Returns a PyIceberg Catalog from the given catalog configuration. """
+        catalog_type = cls.catalog_type_map.get(catalog.type.__class__.__name__, None)
+
+        if not catalog_type:
+            raise NotImplementedError(f"Iceberg Catalog of type ['{catalog.type.__class__.__name__}'] Not Implemmented.")
+
+        return catalog_type.get_catalog(catalog)

ingestion/src/metadata/ingestion/source/database/iceberg/catalog/base.py

@@ -0,0 +1,33 @@
+#  Copyright 2021 Collate
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#  http://www.apache.org/licenses/LICENSE-2.0
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+"""
+Iceberg Catalog base module.
+"""
+
+from abc import ABC, abstractmethod
+
+from pyiceberg.catalog import Catalog
+
+from metadata.generated.schema.entity.services.connections.database.icebergConnection import Catalog as IcebergCatalog
+from metadata.ingestion.source.database.iceberg.fs import FileSystemConfig, IcebergFileSystemFactory
+
+class IcebergCatalogBase(ABC):
+    @classmethod
+    @abstractmethod
+    def get_catalog(cls, catalog: IcebergCatalog) -> Catalog:
+        """ Returns a Catalog for given catalog configuration. """
+        pass
+
+    @staticmethod
+    def get_fs_parameters(fs_config: FileSystemConfig) -> dict:
+        """ Gets the FileSystem parameters based on passed configuration. """
+        return IcebergFileSystemFactory.parse(fs_config)

ingestion/src/metadata/ingestion/source/database/iceberg/catalog/dynamodb.py

@@ -0,0 +1,53 @@
+#  Copyright 2021 Collate
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#  http://www.apache.org/licenses/LICENSE-2.0
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+"""
+Iceberg DynamoDB Catalog
+"""
+from pyiceberg.catalog import Catalog, load_dynamodb
+
+from metadata.generated.schema.entity.services.connections.database.icebergConnection import (
+        Catalog as IcebergCatalog,
+        DynamoDbCatalogConnection,
+)
+from metadata.ingestion.source.database.iceberg.catalog.base import IcebergCatalogBase
+
+class IcebergDynamoDbCatalog(IcebergCatalogBase):
+    @classmethod
+    def get_catalog(cls, catalog: IcebergCatalog) -> Catalog:
+        """ Returns a DynamoDB Catalog for the given connection and file storage.
+
+        For more information, check the PyIceberg [docs](https://py.iceberg.apache.org/configuration/#dynamodb-catalog)
+        """
+        if not isinstance(catalog.type, DynamoDbCatalogConnection):
+            raise RuntimeError("'connection' is not an instance of 'DynamoDbCatalogConnection'")
+
+        parameters = {
+            **cls.get_fs_parameters(catalog.fileSystem),
+            "warehouse": catalog.warehouseLocation
+        }
+
+        if catalog.type.tableName:
+            parameters = {
+                "table-name": catalog.type.tableName
+            }
+
+        if catalog.type.awsConfig:
+            parameters = {
+                **parameters,
+                "aws_secret_key_id": catalog.type.awsConfig.awsAccessKeyId,
+                "aws_secret_access_key": catalog.type.awsConfig.awsSecretAccessKey,
+                "aws_session_token": catalog.type.awsConfig.awsSessionToken,
+                "region_name": catalog.type.awsConfig.awsRegion,
+                "profile_name": catalog.type.awsConfig.profileName,
+            }
+
+        return load_dynamodb(catalog.name, parameters)

ingestion/src/metadata/ingestion/source/database/iceberg/catalog/glue.py

@@ -0,0 +1,48 @@
+#  Copyright 2021 Collate
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#  http://www.apache.org/licenses/LICENSE-2.0
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+"""
+Iceberg Glue Catalog
+"""
+from pyiceberg.catalog import Catalog, load_glue
+
+from metadata.generated.schema.entity.services.connections.database.icebergConnection import (
+    Catalog as IcebergCatalog,
+    GlueCatalogConnection,
+)
+from metadata.ingestion.source.database.iceberg.catalog.base import IcebergCatalogBase
+
+class IcebergGlueCatalog(IcebergCatalogBase):
+    @classmethod
+    def get_catalog(cls, catalog: IcebergCatalog) -> Catalog:
+        """ Returns a Glue Catalog for the given connection and file storage.
+
+        For more information, check the PyIceberg [docs](https://py.iceberg.apache.org/configuration/#glue-catalog)
+        """
+        if not isinstance(catalog.type, GlueCatalogConnection):
+            raise RuntimeError("'connection' is not an instance of 'GlueCatalogConnection'")
+
+        parameters = {
+            **cls.get_fs_parameters(catalog.fileSystem),
+            "warehouse": catalog.warehouseLocation
+        }
+
+        if catalog.type.awsConfig:
+            parameters = {
+                **parameters,
+                "aws_secret_key_id": catalog.type.awsConfig.awsAccessKeyId,
+                "aws_secret_access_key": catalog.type.awsConfig.awsSecretAccessKey,
+                "aws_session_token": catalog.type.awsConfig.awsSessionToken,
+                "region_name": catalog.type.awsConfig.awsRegion,
+                "profile_name": catalog.type.awsConfig.profileName,
+            }
+
+        return load_glue(catalog.name, parameters)

ingestion/src/metadata/ingestion/source/database/iceberg/catalog/hive.py

@@ -0,0 +1,39 @@
+#  Copyright 2021 Collate
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#  http://www.apache.org/licenses/LICENSE-2.0
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+"""
+Iceberg Hive Catalog
+"""
+from pyiceberg.catalog import Catalog, load_hive
+
+from metadata.generated.schema.entity.services.connections.database.icebergConnection import (
+    Catalog as IcebergCatalog,
+    HiveCatalogConnection,
+)
+from metadata.ingestion.source.database.iceberg.catalog.base import IcebergCatalogBase
+
+class IcebergHiveCatalog(IcebergCatalogBase):
+    @classmethod
+    def get_catalog(cls, catalog: IcebergCatalog) -> Catalog:
+        """ Returns a Hive Catalog for the given connection and file storage.
+
+        For more information, check the PyIceberg [docs](https://py.iceberg.apache.org/configuration/#hive-catalog)
+        """
+        if not isinstance(catalog.type, HiveCatalogConnection):
+            raise RuntimeError("'connection' is not an instance of 'HiveCatalogConnection'")
+
+        parameters = {
+            **cls.get_fs_parameters(catalog.fileSystem),
+            "warehouse": catalog.warehouseLocation,
+            "uri": catalog.type.uri,
+        }
+
+        return load_hive(catalog.name, parameters)

ingestion/src/metadata/ingestion/source/database/iceberg/catalog/rest.py

@@ -0,0 +1,60 @@
+#  Copyright 2021 Collate
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#  http://www.apache.org/licenses/LICENSE-2.0
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+"""
+Iceberg Rest Catalog
+"""
+from pyiceberg.catalog import Catalog, load_rest
+
+from metadata.generated.schema.entity.services.connections.database.icebergConnection import (
+    Catalog as IcebergCatalog,
+    RestCatalogConnection,
+)
+from metadata.ingestion.source.database.iceberg.catalog.base import IcebergCatalogBase
+
+class IcebergRestCatalog(IcebergCatalogBase):
+    @classmethod
+    def get_catalog(cls, catalog: IcebergCatalog) -> Catalog:
+        """ Returns a Rest Catalog for the given connection and file storage.
+
+        For more information, check the PyIceberg [docs](https://py.iceberg.apache.org/configuration/#rest-catalog)
+        """
+        if not isinstance(catalog.type, RestCatalogConnection):
+            raise RuntimeError("'connection' is not an instance of 'RestCatalogConnection'")
+
+        parameters = {
+            **cls.get_fs_parameters(catalog.fileSystem),
+            "warehouse": catalog.warehouseLocation,
+            "uri": catalog.type.uri,
+            "credential": catalog.type.credential,
+            "token": catalog.type.token,
+        }
+
+        if catalog.type.ssl:
+            parameters = {
+                **parameters,
+                "ssl": {
+                    "client": {
+                        "cert": catalog.type.ssl.clientCertPath,
+                        "key": catalog.type.ssl.privateKeyPath
+                    },
+                    "cabundle": catalog.type.ssl.caCertPath
+                }
+            }
+
+        if catalog.type.sigv4:
+            parameters = {
+                **parameters,
+                "rest.sigv4": True,
+                "rest.signing_region": catalog.type.sigv4.signingRegion,
+                "rest.signing_name": catalog.type.sigv4.signingName
+            }
+        return load_rest(catalog.name, parameters)