S3, GCS, Azure Blob Destinations

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'
 }
 

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())    

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)

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 """
 

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):

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)
 

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
 

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",
 ]