diff --git a/data-transfer/pontoon/pontoon/__init__.py b/data-transfer/pontoon/pontoon/__init__.py index 1bc452a..f94323c 100644 --- a/data-transfer/pontoon/pontoon/__init__.py +++ b/data-transfer/pontoon/pontoon/__init__.py @@ -8,6 +8,7 @@ from pontoon.destination.bigquery_destination import BigQueryDestination from pontoon.destination.s3_destination import S3Destination from pontoon.destination.gcs_destination import GCSDestination +from pontoon.destination.abs_destination import ABSDestination from pontoon.destination.snowflake_storage_destination import SnowflakeStorageDestination from pontoon.destination.snowflake_destination import SnowflakeDestination from pontoon.destination.postgres_destination import PostgresDestination @@ -59,6 +60,7 @@ def get_destination_by_vendor(vendor_type:str) -> str: __destinations['destination-bigquery'] = BigQueryDestination __destinations['destination-s3'] = S3Destination __destinations['destination-gcs'] = GCSDestination +__destinations['destination-abs'] = ABSDestination __destinations['destination-snowflake-storage'] = SnowflakeStorageDestination __destinations['destination-snowflake'] = SnowflakeDestination __destinations['destination-postgres'] = PostgresDestination @@ -88,6 +90,7 @@ def get_destination_by_vendor(vendor_type:str) -> str: 'glue': 'destination-glue-s3', 's3': 'destination-s3', 'gcs': 'destination-gcs', + 'abs': 'destination-abs', 'postgresql': 'destination-postgres' } diff --git a/data-transfer/pontoon/pontoon/destination/abs_destination.py b/data-transfer/pontoon/pontoon/destination/abs_destination.py new file mode 100644 index 0000000..fb746e1 --- /dev/null +++ b/data-transfer/pontoon/pontoon/destination/abs_destination.py @@ -0,0 +1,99 @@ +import os +from typing import List, Dict, Any +from azure.storage.blob import BlobServiceClient +from pontoon.base import Namespace, Destination, Stream, Dataset, Record, Progress +from pontoon.base import DestinationError +from pontoon.destination import ObjectStoreBase +from pontoon.destination.integrity import ABSIntegrity + + +class ABSConfig: + """ A class to represent an Azure Blob configuration block """ + + def __init__(self, config): + self.scheme = 'blob' + self.bucket_name = config.get('blob_container') + self.bucket_path = config.get('blob_prefix', '') + self.region = '' + + self.bucket_name = self.bucket_name.lstrip('abfss://').rstrip('/') + self.bucket_path = self.bucket_path.lstrip('/').rstrip('/') + + + +class ABSDestination(ObjectStoreBase): + """ A Destination that writes to Azure Blob Store in Parquet format """ + + + def __init__(self, config): + + super().__init__(config) + + # our azure blob config config + self._abs_config = ABSConfig(config['connect']) + + if self._format not in ['staging', 'hive']: + raise DestinationError(f'Format {self._format} is not supported by Azure Blob Store') + + + def _get_abs_client(self): + # get Azure Blob container client using configured auth type + + connect = self._config.get('connect') + auth_type = connect.get('auth_type') + + if auth_type not in ['connection_string']: + raise Exception(f"ABSDestination (destination-abs) does not support auth type '{auth_type}'") + + blob_service_client = BlobServiceClient.from_connection_string( + connect.get('blob_connection_string') + ) + return blob_service_client.get_container_client( + connect.get('blob_container') + ) + + def _write_stream(self, stream:Stream): + pass + + def _write_batch(self, stream:Stream, batch:List[Record], batch_index:int): + + # Write a batch of records to azure blob formatted as Parquet + abs_client = self._get_abs_client() + + # write the parquet file + parquet_file_path = ObjectStoreBase._write_parquet( + stream, + batch, + parquet_config=self._config.get('parquet', {}) + ) + + # upload to azure + if self._format == 'staging': + parquet_abs_path = ObjectStoreBase.get_object_filename( + self._abs_config, + self._ds.namespace, + stream, + self._dt, + self._batch_id, + batch_index + ) + elif self._format == 'hive': + parquet_abs_path = ObjectStoreBase.get_hive_filename( + self._abs_config, + self._ds.namespace, + stream, + self._dt, + self._batch_id, + batch_index + ) + + blob = abs_client.get_blob_client(parquet_abs_path) + with open(parquet_file_path, 'rb') as data: + blob.upload_blob(data, overwrite=True) + + # clean up + os.remove(parquet_file_path) + + + def integrity(self): + return ABSIntegrity(self._get_abs_client()) diff --git a/data-transfer/pontoon/pontoon/destination/gcs_destination.py b/data-transfer/pontoon/pontoon/destination/gcs_destination.py index 0c3fba8..f96ea80 100644 --- a/data-transfer/pontoon/pontoon/destination/gcs_destination.py +++ b/data-transfer/pontoon/pontoon/destination/gcs_destination.py @@ -4,6 +4,7 @@ from typing import List, Dict, Any from google.cloud import storage from pontoon.base import Namespace, Destination, Stream, Dataset, Record, Progress +from pontoon.base import DestinationError from pontoon.destination import ObjectStoreBase from pontoon.destination.integrity import GCSIntegrity @@ -42,7 +43,14 @@ def __init__(self, config): temp_file.write(connect.get('service_account')) self._service_account_file = temp_file.name + if self._format not in ['staging', 'hive']: + raise DestinationError(f'Format {self._format} is not supported by GCS') + + def _write_stream(self, stream:Stream): + pass + + def _write_batch(self, stream:Stream, batch:List[Record], batch_index:int): # Write a batch of records to GCS formatted as Parquet @@ -57,14 +65,24 @@ def _write_batch(self, stream:Stream, batch:List[Record], batch_index:int): ) # upload to GCS - parquet_gcs_path = ObjectStoreBase.get_object_filename( - self._gcs_config, - self._ds.namespace, - stream, - self._dt, - self._batch_id, - batch_index - ) + if self._format == 'staging': + parquet_gcs_path = ObjectStoreBase.get_object_filename( + self._gcs_config, + self._ds.namespace, + stream, + self._dt, + self._batch_id, + batch_index + ) + elif self._format == 'hive': + parquet_gcs_path = ObjectStoreBase.get_hive_filename( + self._gcs_config, + self._ds.namespace, + stream, + self._dt, + self._batch_id, + batch_index + ) blob = bucket.blob(parquet_gcs_path) diff --git a/data-transfer/pontoon/pontoon/destination/integrity.py b/data-transfer/pontoon/pontoon/destination/integrity.py index f5088a6..a228335 100644 --- a/data-transfer/pontoon/pontoon/destination/integrity.py +++ b/data-transfer/pontoon/pontoon/destination/integrity.py @@ -53,6 +53,17 @@ def check_batch_volume(self, ds:Dataset): return True +class ABSIntegrity(Integrity): + """ Integrity checker for Azure Blob destinations """ + + def __init__(self, client): + self._client = client + raise NotImplementedError("ABSIntegrity checker is not implemented yet") + + def check_batch_volume(self, ds:Dataset): + return True + + class SMSIntegrity(Integrity): """ Integrity checker for SMS (Snowflake) based destinations """ diff --git a/data-transfer/pontoon/pontoon/destination/object_store_base.py b/data-transfer/pontoon/pontoon/destination/object_store_base.py index 8f63930..c12c5ba 100644 --- a/data-transfer/pontoon/pontoon/destination/object_store_base.py +++ b/data-transfer/pontoon/pontoon/destination/object_store_base.py @@ -88,17 +88,50 @@ def get_object_path_uri(config:ObjectStoreConfig, namespace:Namespace, stream:St # e.g. s3://bucket/events/postgres/pontoon__events/2025-01-10/1740773449235/ return f"{config.scheme}://{config.bucket_name}/{ObjectStoreBase.get_object_path(config, namespace, stream, dt, batch_id)}" - + + @staticmethod + def get_hive_name(stream:Stream, dt:datetime, batch_id:str, batch_index:int): + # e.g. 20250701154312_1740773449235_0.parquet + return f"{dt.strftime('%Y%m%d%H%M%S')}_{batch_id}_{batch_index}.parquet" + + + @staticmethod + def get_hive_path(config:ObjectStoreConfig, namespace:Namespace, stream:Stream, dt:datetime, batch_id:str): + # e.g. events/events/dt=2025-01-10/ + date = dt.strftime('%Y-%m-%d') + return f"{config.bucket_path}/{stream.name}/dt={date}/" + + + @staticmethod + def get_hive_filename(config:ObjectStoreConfig, namespace:Namespace, stream:Stream, dt:datetime, batch_id:str, batch_index:int): + # e.g. events/events/dt=2025-01-10/20250110154312_1740773449235_0.parquet + return f"{ObjectStoreBase.get_hive_path(config, namespace, stream, dt, batch_id)}{ObjectStoreBase.get_hive_name(stream, dt, batch_id, batch_index)}" + + + @staticmethod + def get_hive_path_uri(config:ObjectStoreConfig, namespace:Namespace, stream:Stream, dt:datetime, batch_id:str): + # e.g. s3://bucket/events/events/dt=2025-01-10/ + return f"{config.scheme}://{config.bucket_name}/{ObjectStoreBase.get_hive_path(config, namespace, stream, dt, batch_id)}" + def __init__(self, config): self._config = config + self._mode = config.get('mode') self._batch_size = config.get('batch_size', 10000) + + # default format is as a staging store for transfers between other stores + # options are staging, hive + self._format = config.get('connect').get('format', 'staging').lower() self._dt = None self._batch_id = None self._ds = None + @abstractmethod + def _write_stream(self, stream:Stream): pass + + @abstractmethod def _write_batch(self, stream:Stream, batch:List[Record], batch_index:int): pass @@ -122,6 +155,8 @@ def write(self, ds:Dataset, progress_callback=None): if callable(progress_callback): progress.subscribe(progress_callback) + self._write_stream(stream) + batch = [] batch_index = 0 for record in ds.read(stream): diff --git a/data-transfer/pontoon/pontoon/destination/s3_destination.py b/data-transfer/pontoon/pontoon/destination/s3_destination.py index 7993b81..9ff2baa 100644 --- a/data-transfer/pontoon/pontoon/destination/s3_destination.py +++ b/data-transfer/pontoon/pontoon/destination/s3_destination.py @@ -2,6 +2,7 @@ from typing import List, Dict, Any import boto3 from pontoon.base import Namespace, Destination, Stream, Dataset, Record, Progress +from pontoon.base import DestinationError from pontoon.destination import ObjectStoreBase from pontoon.destination.integrity import S3Integrity @@ -31,6 +32,8 @@ def __init__(self, config): # our s3 config self._s3_config = S3Config(config['connect']) + if self._format not in ['staging', 'hive']: + raise DestinationError(f'Format {self._format} is not supported by S3') def _get_s3_client(self): # get S3 client using configured auth type @@ -48,6 +51,8 @@ def _get_s3_client(self): region_name=self._s3_config.region ) + def _write_stream(self, stream:Stream): + pass def _write_batch(self, stream:Stream, batch:List[Record], batch_index:int): @@ -62,14 +67,24 @@ def _write_batch(self, stream:Stream, batch:List[Record], batch_index:int): ) # upload to s3 - parquet_s3_path = ObjectStoreBase.get_object_filename( - self._s3_config, - self._ds.namespace, - stream, - self._dt, - self._batch_id, - batch_index - ) + if self._format == 'staging': + parquet_s3_path = ObjectStoreBase.get_object_filename( + self._s3_config, + self._ds.namespace, + stream, + self._dt, + self._batch_id, + batch_index + ) + elif self._format == 'hive': + parquet_s3_path = ObjectStoreBase.get_hive_filename( + self._s3_config, + self._ds.namespace, + stream, + self._dt, + self._batch_id, + batch_index + ) s3.upload_file(parquet_file_path, self._s3_config.bucket_name, parquet_s3_path) diff --git a/data-transfer/pontoon/pontoon/destination/snowflake_storage_destination.py b/data-transfer/pontoon/pontoon/destination/snowflake_storage_destination.py index 096fb30..774a89f 100644 --- a/data-transfer/pontoon/pontoon/destination/snowflake_storage_destination.py +++ b/data-transfer/pontoon/pontoon/destination/snowflake_storage_destination.py @@ -3,7 +3,7 @@ from datetime import datetime import snowflake.connector from pontoon.base import Namespace, Destination, Stream, Dataset, Record, Progress -from pontoon.base import DestinationConnectionFailed +from pontoon.base import DestinationError, DestinationConnectionFailed from pontoon.source.sql_source import SQLUtil from pontoon.destination import ObjectStoreBase from pontoon.destination.integrity import SMSIntegrity @@ -25,6 +25,9 @@ def __init__(self, config): self._create_stage = connect.get('create_stage', False) self._parquet_config = connect.get('parquet', {}) + if self._format != 'staging': + raise DestinationError(f'Format {self._format} is not supported by Snowflake Storage') + def _get_snowflake_client(self): c = self._config.get('connect') @@ -40,7 +43,11 @@ def _get_snowflake_client(self): except Exception as e: raise DestinationConnectionFailed("Failed to connect to Snowflake") from e - + + def _write_stream(self, stream:Stream): + pass + + def _write_batch(self, stream:Stream, batch:List[Record], batch_index:int): # Write a batch of records to snowflake storage formatted as Parquet diff --git a/data-transfer/pontoon/pyproject.toml b/data-transfer/pontoon/pyproject.toml index 295b84a..d7c80dd 100644 --- a/data-transfer/pontoon/pyproject.toml +++ b/data-transfer/pontoon/pyproject.toml @@ -20,6 +20,7 @@ dependencies = [ "pyathena==3.12.2", "google-cloud-storage==2.19.0", "google-cloud-bigquery-storage==2.27.0", + "azure-storage-blob==12.26.0", "celery==5.5.3", "celery-redbeat==2.3.2", ] diff --git a/data-transfer/pontoon/tests/integration/test_abs_connectors.py b/data-transfer/pontoon/tests/integration/test_abs_connectors.py new file mode 100644 index 0000000..675dc0c --- /dev/null +++ b/data-transfer/pontoon/tests/integration/test_abs_connectors.py @@ -0,0 +1,119 @@ +import os +import json +import glob +import uuid +import pytest +from datetime import datetime, timezone + +from sqlalchemy import create_engine, inspect, MetaData, Table, text + +from pontoon import configure_logging +from pontoon import get_source, get_destination, get_source_by_vendor, get_destination_by_vendor +from pontoon import SqliteCache +from pontoon import Progress, Mode +from pontoon import SourceConnectionFailed, SourceStreamDoesNotExist, SourceStreamInvalidSchema +from pontoon import DestinationConnectionFailed, DestinationStreamInvalidSchema + +from tests.integration.common import clear_cache_files, get_memory_source, read_progress_handler, write_progress_handler + +from dotenv import load_dotenv +load_dotenv() + + +clear_cache_files() + +class TestABSConnectors: + + def test_abs_source(self): + + def get_test_source(mode_config={}, with_config={}, connect_config={}, streams_config={}): + test_mode_config = { + 'type': Mode.INCREMENTAL, + 'period': Mode.DAILY, + 'start': datetime(2025, 7, 1, tzinfo=timezone.utc), + 'end': datetime(2025, 7, 2, tzinfo=timezone.utc) + } | mode_config + + test_with_config = { + 'batch_id': True, + 'last_sync': True + } | with_config + + test_connect_config = { + + } | connect_config + + test_streams_config = [{ + 'schema': 'source', + 'table': 'leads_xs', + 'primary_field': 'id', + 'cursor_field': 'updated_at', + 'filters': {'customer_id': 'Customer1'}, + 'drop_fields': ['customer_id'] + } | streams_config] + + return get_source( + get_source_by_vendor('s3'), + config = { + 'mode': Mode(test_mode_config), + 'with': test_with_config, + 'connect': test_connect_config, + 'streams': test_streams_config + }, + cache_implementation = SqliteCache, + cache_config = { + 'db': f"_abs_{uuid.uuid4()}_cache.db", + 'chunk_size': 1024 + } + ) + + # Azure Blob is not available as a source connector yet + return + + def test_abs_destination(self): + + def get_test_destination(mode_config={}, connect_config={}): + + test_mode_config = { + 'type': Mode.INCREMENTAL, + 'period': Mode.DAILY, + 'start': datetime(2025, 1, 1, tzinfo=timezone.utc), + 'end': datetime(2025, 1, 2, tzinfo=timezone.utc) + } | mode_config + + test_connect_config = { + 'auth_type': 'connection_string', + 'vendor_type': 'abs', + 'format': 'HIVE', + 'blob_container': os.environ['AZURE_BLOB_CONTAINER'], + 'blob_prefix': 'pontoon-hive', + 'blob_connection_string': os.environ['AZURE_CONNECTION_STRING'] + } | connect_config + + return get_destination( + get_destination_by_vendor('abs'), + config = { + 'mode': Mode(test_mode_config), + 'connect': test_connect_config + } + ) + + def connect(): + pass + + def drop(): + pass + + + drop() + + # Full refresh + dest = get_test_destination(mode_config={'type': Mode.FULL_REFRESH}) + ds = get_memory_source(mode_config={'type': Mode.FULL_REFRESH}).read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + + # Azure Blob integrity checker is not implemented yet + # dest.integrity().check_batch_volume(ds) + + drop() + diff --git a/data-transfer/pontoon/tests/integration/test_gcs_connectors.py b/data-transfer/pontoon/tests/integration/test_gcs_connectors.py new file mode 100644 index 0000000..6d2c0e7 --- /dev/null +++ b/data-transfer/pontoon/tests/integration/test_gcs_connectors.py @@ -0,0 +1,119 @@ +import os +import json +import glob +import uuid +import pytest +from datetime import datetime, timezone + +from sqlalchemy import create_engine, inspect, MetaData, Table, text + +from pontoon import configure_logging +from pontoon import get_source, get_destination, get_source_by_vendor, get_destination_by_vendor +from pontoon import SqliteCache +from pontoon import Progress, Mode +from pontoon import SourceConnectionFailed, SourceStreamDoesNotExist, SourceStreamInvalidSchema +from pontoon import DestinationConnectionFailed, DestinationStreamInvalidSchema + +from tests.integration.common import clear_cache_files, get_memory_source, read_progress_handler, write_progress_handler + +from dotenv import load_dotenv +load_dotenv() + + +clear_cache_files() + +class TestGCSConnectors: + + def test_gcs_source(self): + + def get_test_source(mode_config={}, with_config={}, connect_config={}, streams_config={}): + test_mode_config = { + 'type': Mode.INCREMENTAL, + 'period': Mode.DAILY, + 'start': datetime(2025, 7, 1, tzinfo=timezone.utc), + 'end': datetime(2025, 7, 2, tzinfo=timezone.utc) + } | mode_config + + test_with_config = { + 'batch_id': True, + 'last_sync': True + } | with_config + + test_connect_config = { + + } | connect_config + + test_streams_config = [{ + 'schema': 'source', + 'table': 'leads_xs', + 'primary_field': 'id', + 'cursor_field': 'updated_at', + 'filters': {'customer_id': 'Customer1'}, + 'drop_fields': ['customer_id'] + } | streams_config] + + return get_source( + get_source_by_vendor('gcs'), + config = { + 'mode': Mode(test_mode_config), + 'with': test_with_config, + 'connect': test_connect_config, + 'streams': test_streams_config + }, + cache_implementation = SqliteCache, + cache_config = { + 'db': f"_gcs_{uuid.uuid4()}_cache.db", + 'chunk_size': 1024 + } + ) + + # S3 is not available as a source connector yet + return + + def test_gcs_destination(self): + + def get_test_destination(mode_config={}, connect_config={}): + + test_mode_config = { + 'type': Mode.INCREMENTAL, + 'period': Mode.DAILY, + 'start': datetime(2025, 1, 1, tzinfo=timezone.utc), + 'end': datetime(2025, 1, 2, tzinfo=timezone.utc) + } | mode_config + + test_connect_config = { + 'auth_type': 'service_account', + 'vendor_type': 'gcs', + 'format': 'HIVE', + 'gcs_bucket_name': os.environ['GCS_BUCKET'], + 'gcs_bucket_path': 'pontoon-hive', + 'service_account': open(os.environ['GCP_SERVICE_ACCOUNT_FILE']).read() + } | connect_config + + return get_destination( + get_destination_by_vendor('gcs'), + config = { + 'mode': Mode(test_mode_config), + 'connect': test_connect_config + } + ) + + def connect(): + pass + + def drop(): + pass + + + drop() + + # Full refresh + dest = get_test_destination(mode_config={'type': Mode.FULL_REFRESH}) + ds = get_memory_source(mode_config={'type': Mode.FULL_REFRESH}).read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + + # GCS integrity checker is not implemented yet + # dest.integrity().check_batch_volume(ds) + + drop() + diff --git a/data-transfer/pontoon/tests/integration/test_s3_connectors.py b/data-transfer/pontoon/tests/integration/test_s3_connectors.py new file mode 100644 index 0000000..ad5540c --- /dev/null +++ b/data-transfer/pontoon/tests/integration/test_s3_connectors.py @@ -0,0 +1,121 @@ +import os +import json +import glob +import uuid +import pytest +from datetime import datetime, timezone + +from sqlalchemy import create_engine, inspect, MetaData, Table, text + +from pontoon import configure_logging +from pontoon import get_source, get_destination, get_source_by_vendor, get_destination_by_vendor +from pontoon import SqliteCache +from pontoon import Progress, Mode +from pontoon import SourceConnectionFailed, SourceStreamDoesNotExist, SourceStreamInvalidSchema +from pontoon import DestinationConnectionFailed, DestinationStreamInvalidSchema + +from tests.integration.common import clear_cache_files, get_memory_source, read_progress_handler, write_progress_handler + +from dotenv import load_dotenv +load_dotenv() + + +clear_cache_files() + +class TestS3Connectors: + + def test_s3_source(self): + + def get_test_source(mode_config={}, with_config={}, connect_config={}, streams_config={}): + test_mode_config = { + 'type': Mode.INCREMENTAL, + 'period': Mode.DAILY, + 'start': datetime(2025, 7, 1, tzinfo=timezone.utc), + 'end': datetime(2025, 7, 2, tzinfo=timezone.utc) + } | mode_config + + test_with_config = { + 'batch_id': True, + 'last_sync': True + } | with_config + + test_connect_config = { + + } | connect_config + + test_streams_config = [{ + 'schema': 'source', + 'table': 'leads_xs', + 'primary_field': 'id', + 'cursor_field': 'updated_at', + 'filters': {'customer_id': 'Customer1'}, + 'drop_fields': ['customer_id'] + } | streams_config] + + return get_source( + get_source_by_vendor('s3'), + config = { + 'mode': Mode(test_mode_config), + 'with': test_with_config, + 'connect': test_connect_config, + 'streams': test_streams_config + }, + cache_implementation = SqliteCache, + cache_config = { + 'db': f"_s3_{uuid.uuid4()}_cache.db", + 'chunk_size': 1024 + } + ) + + # S3 is not available as a source connector yet + return + + def test_s3_destination(self): + + def get_test_destination(mode_config={}, connect_config={}): + + test_mode_config = { + 'type': Mode.INCREMENTAL, + 'period': Mode.DAILY, + 'start': datetime(2025, 1, 1, tzinfo=timezone.utc), + 'end': datetime(2025, 1, 2, tzinfo=timezone.utc) + } | mode_config + + test_connect_config = { + 'auth_type': 'basic', + 'vendor_type': 's3', + 'format': 'HIVE', + 's3_bucket': os.environ['S3_BUCKET'], + 's3_prefix': 'pontoon-hive', + 's3_region': os.environ['AWS_DEFAULT_REGION'], + 'aws_access_key_id': os.environ['AWS_ACCESS_KEY_ID'], + 'aws_secret_access_key': os.environ['AWS_SECRET_ACCESS_KEY'] + } | connect_config + + return get_destination( + get_destination_by_vendor('s3'), + config = { + 'mode': Mode(test_mode_config), + 'connect': test_connect_config + } + ) + + def connect(): + pass + + def drop(): + pass + + + drop() + + # Full refresh + dest = get_test_destination(mode_config={'type': Mode.FULL_REFRESH}) + ds = get_memory_source(mode_config={'type': Mode.FULL_REFRESH}).read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + + # S3 integrity checker is not implemented yet + # dest.integrity().check_batch_volume(ds) + + drop() + diff --git a/docs/docs/index.md b/docs/docs/index.md index 8211b9c..716f80e 100644 --- a/docs/docs/index.md +++ b/docs/docs/index.md @@ -27,7 +27,9 @@ Pontoon is an open source, self-hosted data export platform that is built from t - **Data Warehouses**: Snowflake, Google BigQuery, Amazon Redshift - **SQL Databases**: Postgres -- _Coming Soon - Object Storage_: Amazon S3, Google Cloud Storage +- **Object Storage**: Amazon S3, Google Cloud Storage, Azure Blob Storage +- _Coming Soon - Data Lake Table Formats: Iceberg, Delta Lake and Hudi_ + ### The Problem with APIs & Data diff --git a/docs/docs/sources-destinations/destinations/S3.md b/docs/docs/sources-destinations/destinations/S3.md new file mode 100644 index 0000000..b2049e2 --- /dev/null +++ b/docs/docs/sources-destinations/destinations/S3.md @@ -0,0 +1,51 @@ +# S3 Destination + +Configure [Amazon S3](https://aws.amazon.com/s3/) as a destination for your data transfers in Pontoon. + +## Prerequisites + +Before configuring S3 as a destination, ensure you have: + +- **AWS Account** +- **S3 Bucket**: S3 bucket that data will be sent to +- **Credentials**: AWS Credentials with `s3:PutObject`, `s3:DeleteObject` permissions for your bucket + +## How it works + +The S3 connector writes data to your bucket as compressed **Apache Parquet** files using **Apache Hive** style partitions, which is supported by most query engines and data platforms making it easy to work with. + +This connector is append-only, so re-running syncs will produce new files with a later timestamp and different batch ID (see below) but will not delete existing data in the destination. + +## Structure of landed data + +Data written to S3 will have the following structure: + +`s3://///dt=/__.parquet` + +- `` is the name of your S3 bucket +- `` is an optional folder prefix +- `` is the name of the data model transferred, similar to a table name +- `` is the date that the transfer started in the format `2025-01-01` +- `` is a timestand of when the transfer started in the format `20250301121507` +- `` is a batch ID generated by Pontoon that is unique to the running transfer - subsequent transfers to the same destination will have different batch IDs +- `` is a monotonically increasing integer for a given batch ID + + +## Configuration + +| Parameter | Description | Required | Example | +| ----------------------- | ----------------------------------- | -------- | ---------------------------------------------------- | +| `s3_bucket` | S3 Bucket Name | Yes | `my-bucket` | +| `s3_prefix` | Folder path prefix | Optional | /exports| +| `s3_region` | Region of bucket | Yes | us-east-1 | + `aws_access_key_id` | AWS access key ID | Yes | `AKIAIOSFODNN7EXAMPLE` | +| `aws_secret_access_key` | AWS secret access key | Yes | `wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY` | + +### Setup Destination + +1. Navigate to **Destinations** → **New Destination** +2. Select **S3** as the destination type +3. Enter connection details: +4. Click **Test Connection** to verify +5. Click **Save** to create the destination + diff --git a/docs/docs/sources-destinations/destinations/azure-blob-storage.md b/docs/docs/sources-destinations/destinations/azure-blob-storage.md new file mode 100644 index 0000000..99712b1 --- /dev/null +++ b/docs/docs/sources-destinations/destinations/azure-blob-storage.md @@ -0,0 +1,50 @@ +# Azure Blob Storage Destination + +Configure [Azure Blob Storage](https://azure.microsoft.com/en-us/products/storage/blobs) as a destination for your data transfers in Pontoon. + +## Prerequisites + +Before configuring Azure Blob Storage as a destination, ensure you have: + +- **Azure Account** +- **Azure Storage Account**: An Azure Storage account with **Hierarchical Namespace** enabled +- **Storage Container**: Go to _Storage account > Data storage > Containers_ and create a new container if needed +- **Shared Access Signature (SAS)**: Connection String URI for Azure Storage with permission to read and write Blobs and Files in your container (_Storage account > Security + networking > Shared access signature_) + +## How it works + +The Azure Blob Storage connector writes data to your bucket as compressed **Apache Parquet** files using **Apache Hive** style partitions, which is supported by most query engines and data platforms making it easy to work with. + +This connector is append-only, so re-running syncs will produce new files with a later timestamp and different batch ID (see below) but will not delete existing data in the destination. + +## Structure of landed data + +Data written to Azure will have the following structure: + +`abfss://///dt=/__.parquet` + +- `` is the name of your Azure Storage Container +- `` is an optional folder prefix +- `` is the name of the data model transferred, similar to a table name +- `` is the date that the transfer started in the format `2025-01-01` +- `` is a timestand of when the transfer started in the format `20250301121507` +- `` is a batch ID generated by Pontoon that is unique to the running transfer - subsequent transfers to the same destination will have different batch IDs +- `` is a monotonically increasing integer for a given batch ID + + +## Configuration + +| Parameter | Description | Required | Example | +| ----------------------- | ----------------------------------- | -------- | ---------------------------------------------------- | +| `blob_container` | Azure Storage Container Name | Yes | `my-container` | +| `blob_prefix` | Folder path prefix | Optional | /exports| +| `blob_connection_string` | Azure SAS Connectio URI | Yes | `BlobEndpoint=https://.blob.core.windows.net/` | + +### Setup Destination + +1. Navigate to **Destinations** → **New Destination** +2. Select **S3** as the destination type +3. Enter connection details: +4. Click **Test Connection** to verify +5. Click **Save** to create the destination + diff --git a/docs/docs/sources-destinations/destinations/google-cloud-storage.md b/docs/docs/sources-destinations/destinations/google-cloud-storage.md new file mode 100644 index 0000000..76120c8 --- /dev/null +++ b/docs/docs/sources-destinations/destinations/google-cloud-storage.md @@ -0,0 +1,49 @@ +# Google Cloud Storage Destination + +Configure [Google Cloud Storage](https://cloud.google.com/storage) as a destination for your data transfers in Pontoon. + +## Prerequisites + +Before configuring Google Cloud Storage as a destination, ensure you have: + +- **GCP Account** +- **GCS Bucket**: GCS bucket that data will be sent to +- **Service Account**: GCP Service Account credentials permission to read and write to your bucket + +## How it works + +The GCS connector writes data to your bucket as compressed **Apache Parquet** files using **Apache Hive** style partitions, which is supported by most query engines and data platforms making it easy to work with. + +This connector is append-only, so re-running syncs will produce new files with a later timestamp and different batch ID (see below) but will not delete existing data in the destination. + +## Structure of landed data + +Data written to GCS will have the following structure: + +`gs://///dt=/__.parquet` + +- `` is the name of your GCS bucket +- `` is an optional folder prefix +- `` is the name of the data model transferred, similar to a table name +- `` is the date that the transfer started in the format `2025-01-01` +- `` is a timestand of when the transfer started in the format `20250301121507` +- `` is a batch ID generated by Pontoon that is unique to the running transfer - subsequent transfers to the same destination will have different batch IDs +- `` is a monotonically increasing integer for a given batch ID + + +## Configuration + +| Parameter | Description | Required | Example | +| ----------------------- | ----------------------------------- | -------- | ---------------------------------------------------- | +| `gcs_bucket_name` | GCS Bucket Name | Yes | `my-bucket` | +| `gcs_bucket_path` | Folder path prefix | Optional | /exports| +| `service_account` | Service account credentials | Yes | JSON file | + +### Setup Destination + +1. Navigate to **Destinations** → **New Destination** +2. Select **S3** as the destination type +3. Enter connection details: +4. Click **Test Connection** to verify +5. Click **Save** to create the destination + diff --git a/docs/docs/sources-destinations/overview.md b/docs/docs/sources-destinations/overview.md index f3de9e7..21a7838 100644 --- a/docs/docs/sources-destinations/overview.md +++ b/docs/docs/sources-destinations/overview.md @@ -17,4 +17,5 @@ Pontoon can write data to the following destinations: - **Data Warehouses**: [Snowflake](destinations/snowflake.md), [Amazon Redshift](destinations/redshift.md), [Google BigQuery](destinations/bigquery.md) - **SQL Databases**: [Postgres](destinations/postgresql.md) -- _Coming Soon - Object Storage_: Amazon S3, Google Cloud Storage +- **Object Storage**: [Amazon S3](https://aws.amazon.com/s3/), [Google Cloud Storage](https://cloud.google.com/storage), [Azure Blob Storage](https://azure.microsoft.com/en-us/products/storage/blobs) +- _Coming Soon - Data Lake Table Formats: Iceberg, Delta Lake and Hudi_ diff --git a/docs/mkdocs.yml b/docs/mkdocs.yml index d5251ca..0857110 100644 --- a/docs/mkdocs.yml +++ b/docs/mkdocs.yml @@ -98,3 +98,6 @@ nav: - Amazon Redshift: sources-destinations/destinations/redshift.md - Google BigQuery: sources-destinations/destinations/bigquery.md - Postgres: sources-destinations/destinations/postgresql.md + - S3: sources-destinations/destinations/S3.md + - Google Cloud Storage: sources-destinations/destinations/google-cloud-storage.md + - Azure Blob Storage: sources-destinations/destinations/azure-blob-storage.md