Metadata-Version: 2.1
Name: datalakesurfer
Version: 0.1.2
Summary: A Python package for Azure Datalake Storage adlsgen2 [abfss://] and Microsoft Fabric Lakehouse [abfss://], Google Cloud Storage [gs://bucket], AWS S3 bucket [s3://bucket] enables format detection and schema retrieval for Iceberg, Delta, and Parquet formats.It also helps identify paritioned columns for parquet datasets.
Home-page: https://github.com/keshavksingh/datalakesurfer
Author: Keshav Kant Singh
Author-email: masterkeshav@gmail.com
Classifier: Programming Language :: Python :: 3
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Requires-Python: >=3.8
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: pyarrowfs-adlgen2==0.2.5
Requires-Dist: adlfs==2024.12.0
Requires-Dist: python-dotenv==1.0.1
Requires-Dist: azure-identity==1.17.1
Requires-Dist: azure-storage-file-datalake==12.21.0
Requires-Dist: numpy==1.26.4
Requires-Dist: pandas==1.5.3
Requires-Dist: pytest==8.3.3
Requires-Dist: pyOpenSSL==24.2.1
Requires-Dist: cryptography==43.0.1
Requires-Dist: duckdb==1.3.2
Requires-Dist: deltalake==1.1.4
Requires-Dist: pydantic==2.11.7
Requires-Dist: gcsfs==2025.7.0
Requires-Dist: s3fs==2025.7.0

# DataLakeSurfer

> **A unified toolkit for detecting, exploring, and validating data lake formats** across **Azure Data Lake Storage Gen2** , **Microsoft Fabric Lakehouse**, **GCP Google Storage** and **AWS S3**.

DataLakeSurfer lets you:

- Detect whether a directory is **Iceberg**, **Delta**, or **Parquet**.
- Retrieve **normalized schema** for supported formats.
- Detect **partition columns** for Parquet datasets.
- Validate configuration inputs with **Pydantic models**.
- Work seamlessly with Azure **ADLS Gen2** and **Fabric Lakehouse**, **AWS S3** and **GCP GS**.

---

## ðŸ“‘ Table of Contents

1. [Installation](#-installation)
2. [Quick Start](#-quick-start)
3. [Core Concepts](#-core-concepts)
4. [Usage Examples](#-usage-examples)
   - [1. Detect Format (ADLS)](#1-detect-format-adls)
   - [2. Detect Format (Fabric)](#2-detect-format-fabric)
   - [3. Detect Format (GCS)](#3-detect-format-gcs)
   - [4. Retrieve Schema](#4-retrieve-schema)
   - [5. Detect Partitions](#5-detect-partitions)
   - [6. AWS S3 Usage](#6-aws-s3-usage)
5. [Supported Formats](#-supported-formats)
6. [Development & Testing](#-development--testing)
7. [License](#-license)

---

## ðŸš€ Installation

```bash
pip install datalakesurfer
```

Requirements:

- Python 3.9+
- `pyarrowfs-adlgen2==0.2.5`
- `adlfs==2024.12.0`
- `azure-identity==1.17.1`
- `azure-storage-file-datalake==12.21.0`
- `numpy==1.26.4`
- `pandas==1.5.3`
- `cryptography==43.0.1`
- `duckdb==1.3.2`
- `deltalake==1.1.4`
- `pydantic==2.11.7`
- `gcsfs==2025.7.0`
- `s3fs==2025.7.0`

---

## âš¡ Quick Start

### Generate Token

```python
from azure.identity import DefaultAzureCredential, ClientSecretCredential
from azure.core.credentials import AccessToken

class CustomTokenCredential():
    def __init__(self):
        self.credential = DefaultAzureCredential()

    def get_token(self):
        token_response = self.credential.get_token("https://storage.azure.com/.default")

        return (token_response.token, token_response.expires_on)

class CustomTokenCredentialSeconday(object):
    def __init__(self,token,expires_on):
        self.token = token
        self.expires_on = expires_on

    def get_token(self, *scopes, **kwargs):
        return AccessToken(self.token, self.expires_on)

token,expires_on = CustomTokenCredential().get_token()
print(token)
print(expires_on)
```

```python
from datalakesurfer.format_detector import FormatDetector
f = FormatDetector(account_name="adlssynapseeus01",
    file_system_name="datacontainer",
    directory_path="salesorder",
    token=token,
    expires_on=expires_on).detect_format()
print(f)

# â†’ {'status': 'success', 'format': 'delta'}
```

---

## ðŸ§  Core Concepts

- **Format Detectors**: Identify dataset format (`iceberg`, `delta`, `parquet`).
- **Schema Retrievers**: Extract a **normalized schema** regardless of source format.
- **Partition Detectors**: Identify partitioning and list partition columns.
- **Models**: All incoming parameters validated by **Pydantic** for correctness.

---

## ðŸ“š Usage Examples

### 1. Detect Format (ADLS)

```python
from datalakesurfer.format_detector import FormatDetector
f = FormatDetector(account_name="adlssynapseeus01",
    file_system_name="iceberg-container",
    directory_path="customerdw/mydb/product",
    token=token,
    expires_on=expires_on).detect_format()
print(f)
# {'status': 'success', 'format': 'iceberg'}

from datalakesurfer.format_detector import FormatDetector
f = FormatDetector(account_name="adlssynapseeus01",
    file_system_name="datacontainer",
    directory_path="SaleParquetData",
    token=token,
    expires_on=expires_on).detect_format()
print(f)
# {'status': 'success', 'format': 'parquet'}
```

---

### 2. Detect Format (Fabric)

```python
from datalakesurfer.fabric_format_detector import FabricFormatDetector
f = FabricFormatDetector(account_name="onelake",
    file_system_name="devworkspace",
    directory_path="devlakehouse.Lakehouse/Files/salesorder",
    token=token,
    expires_on=expires_on).detect_format()
print(f)

#abfss://devworkspace@onelake.dfs.fabric.microsoft.com/devlakehouse.Lakehouse/Files/salesorder
# {'status': 'success', 'format': 'delta'}
```

---

### 3. Detect Format (GCS)

```python
import json
with open("/Workspace/Users/<>/fresh-replica-393006-0398d554d3fa.json", "r") as f:
    service_account_dict = json.load(f)

from datalakesurfer.gcs_format_detector import GCSFormatDetector
f = GCSFormatDetector(service_account_info=service_account_dict,
    file_system_name="storagebucket0001",
    directory_path="Country").detect_format()
print(f)
# {'status': 'success', 'format': 'delta'}

f = GCSFormatDetector(service_account_info=service_account_dict,
    file_system_name="storagebucket0001",
    directory_path="ParquetDataSource").detect_format()
print(f)
# {'status': 'success', 'format': 'parquet'}

f = GCSFormatDetector(service_account_info=service_account_dict,
    file_system_name="storagebucket0001",
    directory_path="customerdw/mydb/product").detect_format()
print(f)
# {'status': 'success', 'format': 'iceberg'}
```

---

### 4. Retrieve Schema

For **Parquet/Delta/Iceberg** in ADLS:

```python
from datalakesurfer.schemas.delta_schema import DeltaSchemaRetriever
s = DeltaSchemaRetriever(account_name="adlssynapseeus01",
    file_system_name="datacontainer",
    directory_path="salesorder",
    token=token,
    expires_on=expires_on).get_schema()
print(s)
#{'status': 'success', 'schema': [{'column_name': 'SalesId', 'dtype': 'string'}, {'column_name': 'ProductName', 'dtype': 'string'}, {'column_name': 'SalesDateTime', 'dtype': 'timestamp'}, {'column_name': 'SalesAmount', 'dtype': 'int'}, {'column_name': 'EventProcessingTime', 'dtype': 'timestamp'}]}

from datalakesurfer.schemas.iceberg_schema import IcebergSchemaRetriever
s = IcebergSchemaRetriever(account_name="adlssynapseeus01",
    file_system_name="iceberg-container",
    directory_path="customerdw/mydb/product",
    token=token,
    expires_on=expires_on).get_schema()
print(s)
#{'status': 'success', 'schema': [{'column_name': 'productId', 'dtype': 'string'}, {'column_name': 'productName', 'dtype': 'string'}, {'column_name': 'productDescription', 'dtype': 'string'}, {'column_name': 'productPrice', 'dtype': 'double'}, {'column_name': 'dataModified', 'dtype': 'timestamp'}]}

from datalakesurfer.schemas.parquet_schema import ParquetSchemaRetriever
s = ParquetSchemaRetriever(account_name="adlssynapseeus01",
    file_system_name="datacontainer",
    directory_path="SaleParquetData",
    token=token,
    expires_on=expires_on).get_schema()
print(s)

#{'status': 'success', 'schema': [{'column_name': 'SalesId', 'dtype': 'string'}, {'column_name': 'ProductName', 'dtype': 'string'}, {'column_name': 'SalesDateTime', 'dtype': 'timestamp'}, {'column_name': 'SalesAmount', 'dtype': 'int'}, {'column_name': 'EventProcessingTime', 'dtype': 'timestamp'}]}
```

---

### 4. Retrieve Schema (GCS)

For **Delta/Iceberg/Parquet** in GCS:

```python
from datalakesurfer.schemas.gcs_delta_schema import GCSDeltaSchemaRetriever
s = GCSDeltaSchemaRetriever(service_account_info=service_account_dict,
    file_system_name="storagebucket0001",
    directory_path="Country").get_schema()
print(s)
# [{'column_name': 'SalesId', 'dtype': 'string'}, ...]

from datalakesurfer.schemas.gcs_iceberg_schema import GCSIcebergSchemaRetriever
s = GCSIcebergSchemaRetriever(service_account_info=service_account_dict,
    file_system_name="storagebucket0001",
    directory_path="customerdw/mydb/product").get_schema()
print(s)
# [{'column_name': 'productId', 'dtype': 'string'}, ...]

from datalakesurfer.schemas.gcs_parquet_schema import GCSParquetSchemaRetriever
s = GCSParquetSchemaRetriever(service_account_info=service_account_dict,
    file_system_name="storagebucket0001",
    directory_path="ParquetDataSource").get_schema()
print(s)
# [{'column_name': 'SalesId', 'dtype': 'string'}, ...]
```

---

### 5. Detect Partitions

ADLS:

```python
from datalakesurfer.schemas.parquet_schema import ParquetSchemaRetriever
f = ParquetSchemaRetriever(account_name="adlssynapseeus01",
    file_system_name="datacontainer",
    directory_path="LargeSaleParquetData2Billion",
    token=token,
    expires_on=expires_on).detect_partitions()
print(f)
#{'status': 'success', 'isPartitioned': True, 'partition_columns': [{'column_name': 'ProductName', 'dtype': 'string'}, {'column_name': 'SalesAmount', 'dtype': 'int32'}]}

from datalakesurfer.schemas.parquet_schema import ParquetSchemaRetriever
f = ParquetSchemaRetriever(account_name="adlssynapseeus01",
    file_system_name="datacontainer",
    directory_path="SaleParquetData",
    token=token,
    expires_on=expires_on).detect_partitions()
print(f)
#{'status': 'success', 'isPartitioned': False, 'partition_columns': []}
```

Fabric:

```python
from datalakesurfer.fabric_format_detector import FabricFormatDetector
f = FabricFormatDetector(account_name="onelake",
    file_system_name="devworkspace",
    directory_path="devlakehouse.Lakehouse/Files/Product",
    token=token,
    expires_on=expires_on).detect_format()
print(f)
#{'status': 'success', 'format': 'parquet'}

from datalakesurfer.fabric_format_detector import FabricFormatDetector
f = FabricFormatDetector(account_name="onelake",
    file_system_name="devworkspace",
    directory_path="devlakehouse.Lakehouse/Files/FlightTripData",
    token=token,
    expires_on=expires_on).detect_format()
print(f)
#{'status': 'success', 'format': 'iceberg'}

from datalakesurfer.schemas.fabric_delta_schema import FabricDeltaSchemaRetriever
s = FabricDeltaSchemaRetriever(account_name="onelake",
    file_system_name="devworkspace",
    directory_path="devlakehouse.Lakehouse/Files/salesorder",
    token=token,
    expires_on=expires_on).get_schema()
print(s)
#{'status': 'success', 'schema': [{'column_name': 'SalesId', 'dtype': 'string'}, {'column_name': 'ProductName', 'dtype': 'string'}, {'column_name': 'SalesDateTime', 'dtype': 'timestamp'}, {'column_name': 'SalesAmount', 'dtype': 'int'}, {'column_name': 'EventProcessingTime', 'dtype': 'timestamp'}]}

from datalakesurfer.schemas.fabric_iceberg_schema import FabricIcebergSchemaRetriever
s = FabricIcebergSchemaRetriever(account_name="onelake",
    file_system_name="devworkspace",
    directory_path="devlakehouse.Lakehouse/Files/FlightTripData",
    token=token,
    expires_on=expires_on).get_schema()
print(s)
#{'status': 'success', 'schema': [{'column_name': 'flight_id', 'dtype': 'int'}, {'column_name': 'airline_code', 'dtype': 'int'}, {'column_name': 'passenger_count', 'dtype': 'int'}, {'column_name': 'flight_duration_minutes', 'dtype': 'bigint'}, {'column_name': 'baggage_weight', 'dtype': 'float'}, {'column_name': 'ticket_price', 'dtype': 'double'}, {'column_name': 'tax_amount', 'dtype': 'decimal(38,18)', 'typeproperties': {'precision': 10, 'scale': 2}}, {'column_name': 'destination', 'dtype': 'string'}, {'column_name': 'flight_data', 'dtype': 'string'}, {'column_name': 'is_international', 'dtype': 'boolean'}, {'column_name': 'departure_timestamp', 'dtype': 'timestamp'}, {'column_name': 'arrival_timestamp_ntz', 'dtype': 'timestamp'}, {'column_name': 'flight_date', 'dtype': 'date'}]}

from datalakesurfer.schemas.fabric_parquet_schema import FabricParquetSchemaRetriever
s = FabricParquetSchemaRetriever(account_name="onelake",
    file_system_name="devworkspace",
    directory_path="devlakehouse.Lakehouse/Files/Product",
    token=token,
    expires_on=expires_on).get_schema()
print(s)

# {'status': 'success', 'schema': [{'column_name': 'name', 'dtype': 'string'}, {'column_name': 'age', 'dtype': 'long'}]}

from datalakesurfer.schemas.fabric_parquet_schema import FabricParquetSchemaRetriever
s = FabricParquetSchemaRetriever(account_name="onelake",
    file_system_name="devworkspace",
    directory_path="devlakehouse.Lakehouse/Files/Product",
    token=token,
    expires_on=expires_on).detect_partitions()
print(s)

#{'status': 'success', 'isPartitioned': False, 'partition_columns': []}

from datalakesurfer.schemas.fabric_parquet_schema import FabricParquetSchemaRetriever
f = FabricParquetSchemaRetriever(account_name="onelake",
    file_system_name="devworkspace",
    directory_path="devlakehouse.Lakehouse/Files/Customer",
    token=token,
    expires_on=expires_on).detect_partitions()
print(f)

#{'status': 'success', 'isPartitioned': True, 'partition_columns': [{'column_name': 'city', 'dtype': 'string'}]}
```

---

### 5. Detect Partitions (GCS)

```python
from datalakesurfer.schemas.gcs_parquet_schema import GCSParquetSchemaRetriever
f = GCSParquetSchemaRetriever(service_account_info=service_account_dict,
    file_system_name="storagebucket0001",
    directory_path="ParquetDataSource").detect_partitions()
print(f)
# {'status': 'success', 'isPartitioned': True, 'partition_columns': [{'column_name': 'ProductName', 'dtype': 'string'}, ...]}

f = GCSParquetSchemaRetriever(service_account_info=service_account_dict,
    file_system_name="storagebucket0001",
    directory_path="ParquetDataSource").detect_partitions()
print(f)
# {'status': 'success', 'isPartitioned': False, 'partition_columns': []}
```

---

### 6. AWS S3 Usage

Set your AWS credentials:

```python
AWS_ACCESS_KEY_ID = "<YOUR_AWS_ACCESS_KEY_ID>"
AWS_SECRET_ACCESS_KEY = "<YOUR_AWS_SECRET_ACCESS_KEY>"
AWS_REGION = "us-east-1"
```

#### Detect Format (S3)

```python
from datalakesurfer.s3_format_detector import S3FormatDetector
f = S3FormatDetector(aws_access_key_id=AWS_ACCESS_KEY_ID,
    aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
    aws_region=AWS_REGION,
    file_system_name="devs3bucket001",
    directory_path="Country").detect_format()
print(f)
# {'status': 'success', 'format': 'delta'}

f = S3FormatDetector(aws_access_key_id=AWS_ACCESS_KEY_ID,
    aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
    aws_region=AWS_REGION,
    file_system_name="devs3bucket001",
    directory_path="ParquetDataSource").detect_format()
print(f)
# {'status': 'success', 'format': 'parquet'}

f = S3FormatDetector(aws_access_key_id=AWS_ACCESS_KEY_ID,
    aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
    aws_region=AWS_REGION,
    file_system_name="devs3bucket001",
    directory_path="customerdw/mydb/product").detect_format()
print(f)
# {'status': 'success', 'format': 'iceberg'}
```

#### Retrieve Schema (S3)

```python
from datalakesurfer.schemas.s3_delta_schema import S3DeltaSchemaRetriever
s = S3DeltaSchemaRetriever(aws_access_key_id=AWS_ACCESS_KEY_ID,
    aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
    aws_region=AWS_REGION,
    file_system_name="devs3bucket001",
    directory_path="Country").get_schema()
print(s)
# [{'column_name': 'SalesId', 'dtype': 'string'}, ...]

from datalakesurfer.schemas.s3_iceberg_schema import S3IcebergSchemaRetriever
s = S3IcebergSchemaRetriever(aws_access_key_id=AWS_ACCESS_KEY_ID,
    aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
    aws_region=AWS_REGION,
    file_system_name="devs3bucket001",
    directory_path="customerdw/mydb/product").get_schema()
print(s)
# [{'column_name': 'productId', 'dtype': 'string'}, ...]

from datalakesurfer.schemas.s3_parquet_schema import S3ParquetSchemaRetriever
s = S3ParquetSchemaRetriever(aws_access_key_id=AWS_ACCESS_KEY_ID,
    aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
    aws_region=AWS_REGION,
    file_system_name="devs3bucket001",
    directory_path="ParquetDataSource").get_schema()
print(s)
# [{'column_name': 'SalesId', 'dtype': 'string'}, ...]
```

#### Detect Partitions (S3 Parquet)

```python
from datalakesurfer.schemas.s3_parquet_schema import S3ParquetSchemaRetriever
f = S3ParquetSchemaRetriever(aws_access_key_id=AWS_ACCESS_KEY_ID,
    aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
    aws_region=AWS_REGION,
    file_system_name="devs3bucket001",
    directory_path="ParquetDataSource").detect_partitions()
print(f)
# {'status': 'success', 'isPartitioned': True, 'partition_columns': [{'column_name': 'ProductName', 'dtype': 'string'}, ...]}

f = S3ParquetSchemaRetriever(aws_access_key_id=AWS_ACCESS_KEY_ID,
    aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
    aws_region=AWS_REGION,
    file_system_name="devs3bucket001",
    directory_path="ParquetDataSource").detect_partitions()
print(f)
# {'status': 'success', 'isPartitioned': False, 'partition_columns': []}
```

---

## ðŸ“¦ Supported Formats

| Format  | ADLS Gen2 (Azure) | Fabric OneLake (Azure) | GCS (GCP) | S3 (AWS) |
| ------- | ----------------- | ---------------------- | --------- | -------- |
| Parquet | .                 | .                      | .         | .        |
| Delta   | .                 | .                      | .         | .        |
| Iceberg | .                 | .                      | .         | .        |

---

## ðŸ›  Development & Testing

Run tests:

```bash
pytest
```

---

## ðŸ“„ License

MIT License â€“ See [LICENSE](LICENSE) for details.

---
