Delta Sharing

Delta Sharing Skill

Overview

Delta Sharing is an open protocol for secure data sharing across organizations. This skill covers share configuration, recipient management, access control, and usage monitoring.

Key Benefits:

• Secure cross-organization sharing
• No data duplication
• Real-time data access
• Centralized access control
• Usage tracking and auditing
• Open standard (works with any platform)

When to Use This Skill

Use Delta Sharing when you need to:

• Share data with external partners
• Enable cross-workspace data access
• Distribute data products to consumers
• Implement multi-tenant data access
• Track data consumption patterns
• Maintain centralized governance

Core Concepts

1. Share Configuration

Create and Configure Shares:

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

# Create share
share = w.shares.create(
    name="partner_analytics_share",
    comment="Analytics data for external partners"
)

# Add tables to share
w.shares.update(
    name="partner_analytics_share",
    updates=[
        {
            "action": "ADD",
            "data_object": {
                "name": "catalog.schema.customer_metrics",
                "data_object_type": "TABLE",
                "comment": "Aggregated customer metrics"
            }
        },
        {
            "action": "ADD",
            "data_object": {
                "name": "catalog.schema.product_sales",
                "data_object_type": "TABLE"
            }
        }
    ]
)

2. Recipient Management

Create and Manage Recipients:

# Create recipient
recipient = w.recipients.create(
    name="acme_corp",
    comment="ACME Corporation partner",
    authentication_type="TOKEN"
)

# Get activation URL for recipient
activation_url = recipient.activation_url
print(f"Share this URL with recipient: {activation_url}")

# Grant access to share
w.grants.update(
    securable_type="SHARE",
    securable_name="partner_analytics_share",
    changes=[{
        "principal": "acme_corp",
        "add": ["SELECT"]
    }]
)

3. Access Control

Manage Permissions:

# List current permissions
permissions = w.grants.get_effective(
    securable_type="SHARE",
    full_name="partner_analytics_share"
)

# Revoke access
w.grants.update(
    securable_type="SHARE",
    securable_name="partner_analytics_share",
    changes=[{
        "principal": "acme_corp",
        "remove": ["SELECT"]
    }]
)

# Grant access to specific tables only
w.grants.update(
    securable_type="TABLE",
    full_name="catalog.schema.customer_metrics",
    securable_type="SHARE",
    securable_name="partner_analytics_share",
    changes=[{
        "principal": "acme_corp",
        "add": ["SELECT"]
    }]
)

4. Usage Monitoring

Track Share Usage:

def monitor_share_usage(spark, share_name: str, days: int = 7):
    """Monitor Delta Sharing usage."""
    query = f"""
    SELECT
        date_trunc('day', request_time) as date,
        recipient_name,
        table_name,
        COUNT(*) as request_count,
        SUM(rows_returned) as total_rows,
        SUM(bytes_transferred) as total_bytes
    FROM system.access.audit
    WHERE share_name = '{share_name}'
      AND request_time >= current_date() - INTERVAL {days} DAYS
    GROUP BY 1, 2, 3
    ORDER BY 1 DESC, 4 DESC
    """

    return spark.sql(query)


# Get usage metrics
usage_df = monitor_share_usage(spark, "partner_analytics_share")
usage_df.show()

Implementation Patterns

Pattern 1: Multi-Tenant Sharing

Separate Shares per Tenant:

class MultiTenantSharingManager:
    """Manage multi-tenant data sharing."""

    def __init__(self, workspace_client: WorkspaceClient):
        self.client = workspace_client

    def setup_tenant_share(
        self,
        tenant_name: str,
        tables: List[str],
        tenant_email: str
    ):
        """Set up complete sharing for a tenant."""
        share_name = f"{tenant_name}_share"

        # Create share
        self.client.shares.create(
            name=share_name,
            comment=f"Data share for {tenant_name}"
        )

        # Add tables
        for table in tables:
            self.client.shares.update(
                name=share_name,
                updates=[{
                    "action": "ADD",
                    "data_object": {
                        "name": table,
                        "data_object_type": "TABLE"
                    }
                }]
            )

        # Create recipient
        recipient = self.client.recipients.create(
            name=tenant_name,
            comment=f"Recipient for {tenant_name}",
            authentication_type="TOKEN"
        )

        # Grant access
        self.client.grants.update(
            securable_type="SHARE",
            securable_name=share_name,
            changes=[{
                "principal": tenant_name,
                "add": ["SELECT"]
            }]
        )

        return {
            "share_name": share_name,
            "recipient": tenant_name,
            "activation_url": recipient.activation_url,
            "tables": tables
        }

Pattern 2: Row-Level Security

Implement RLS for Shared Data:

import dlt
from pyspark.sql.functions import *


@dlt.table(
    name="customer_data_with_rls",
    comment="Customer data with row-level security"
)
def customer_data_with_tenant_filter():
    """
    Apply row-level security for multi-tenant sharing.

    Each recipient sees only their tenant's data.
    """
    return (
        dlt.read("silver_customer_data")
        .withColumn("tenant_id", col("customer_id").substr(1, 3))
        .select(
            "customer_id",
            "tenant_id",
            "customer_name",
            "email",
            "created_date"
        )
    )


# Share with partition filtering
def create_tenant_specific_share(
    tenant_id: str,
    source_table: str,
    share_name: str
):
    """
    Create share with automatic tenant filtering.

    Uses partition pruning for efficient filtering.
    """
    w = WorkspaceClient()

    # Create view with tenant filter
    view_name = f"{source_table}_{tenant_id}_view"

    spark.sql(f"""
    CREATE OR REPLACE VIEW {view_name} AS
    SELECT * FROM {source_table}
    WHERE tenant_id = '{tenant_id}'
    """)

    # Share the view
    w.shares.create(name=share_name)
    w.shares.update(
        name=share_name,
        updates=[{
            "action": "ADD",
            "data_object": {
                "name": view_name,
                "data_object_type": "VIEW"
            }
        }]
    )

Pattern 3: Usage Analytics

Comprehensive Usage Tracking:

class SharingAnalytics:
    """Analyze Delta Sharing usage patterns."""

    def __init__(self, spark):
        self.spark = spark

    def get_most_accessed_tables(self, days: int = 30):
        """Find most accessed shared tables."""
        return self.spark.sql(f"""
        SELECT
            table_name,
            COUNT(DISTINCT recipient_name) as unique_recipients,
            COUNT(*) as total_requests,
            SUM(rows_returned) as total_rows,
            AVG(query_duration_ms) as avg_duration_ms
        FROM system.access.audit
        WHERE event_type = 'deltaSharing'
          AND request_time >= current_date() - INTERVAL {days} DAYS
        GROUP BY table_name
        ORDER BY total_requests DESC
        LIMIT 20
        """)

    def get_recipient_activity(self, recipient_name: str, days: int = 7):
        """Detailed activity for specific recipient."""
        return self.spark.sql(f"""
        SELECT
            date_trunc('hour', request_time) as hour,
            table_name,
            COUNT(*) as requests,
            SUM(rows_returned) as rows,
            SUM(bytes_transferred) / 1024 / 1024 as mb_transferred
        FROM system.access.audit
        WHERE recipient_name = '{recipient_name}'
          AND request_time >= current_date() - INTERVAL {days} DAYS
        GROUP BY 1, 2
        ORDER BY 1 DESC
        """)

    def detect_anomalies(self, share_name: str):
        """Detect unusual access patterns."""
        return self.spark.sql(f"""
        WITH daily_stats AS (
            SELECT
                date_trunc('day', request_time) as date,
                recipient_name,
                COUNT(*) as daily_requests
            FROM system.access.audit
            WHERE share_name = '{share_name}'
              AND request_time >= current_date() - INTERVAL 30 DAYS
            GROUP BY 1, 2
        ),
        stats AS (
            SELECT
                recipient_name,
                AVG(daily_requests) as avg_requests,
                STDDEV(daily_requests) as stddev_requests
            FROM daily_stats
            GROUP BY recipient_name
        )
        SELECT
            d.date,
            d.recipient_name,
            d.daily_requests,
            s.avg_requests,
            s.stddev_requests,
            CASE
                WHEN d.daily_requests > s.avg_requests + 2 * s.stddev_requests
                THEN 'ANOMALY_HIGH'
                WHEN d.daily_requests < s.avg_requests - 2 * s.stddev_requests
                THEN 'ANOMALY_LOW'
                ELSE 'NORMAL'
            END as status
        FROM daily_stats d
        JOIN stats s ON d.recipient_name = s.recipient_name
        WHERE date >= current_date() - INTERVAL 7 DAYS
        ORDER BY d.date DESC, d.daily_requests DESC
        """)

Pattern 4: Automated Share Management

CI/CD for Share Configuration:

"""
Automated share management from configuration.
"""
import yaml
from typing import Dict, List, Any


class ShareConfigManager:
    """Manage shares from YAML configuration."""

    def __init__(self, workspace_client: WorkspaceClient):
        self.client = workspace_client

    def load_config(self, config_path: str) -> Dict[str, Any]:
        """Load share configuration from YAML."""
        with open(config_path, 'r') as f:
            return yaml.safe_load(f)

    def apply_config(self, config: Dict[str, Any]):
        """Apply share configuration."""
        for share_config in config.get('shares', []):
            self._create_or_update_share(share_config)

        for recipient_config in config.get('recipients', []):
            self._create_or_update_recipient(recipient_config)

        for grant_config in config.get('grants', []):
            self._apply_grants(grant_config)

    def _create_or_update_share(self, config: Dict[str, Any]):
        """Create or update a share."""
        try:
            existing = self.client.shares.get(name=config['name'])
            print(f"Share {config['name']} already exists")
        except:
            self.client.shares.create(
                name=config['name'],
                comment=config.get('comment', '')
            )
            print(f"Created share: {config['name']}")

        # Add tables
        for table in config.get('tables', []):
            self.client.shares.update(
                name=config['name'],
                updates=[{
                    "action": "ADD",
                    "data_object": {
                        "name": table,
                        "data_object_type": "TABLE"
                    }
                }]
            )

Example Configuration (shares.yaml):

shares:
  - name: partner_analytics_share
    comment: Analytics data for partners
    tables:
      - catalog.analytics.customer_metrics
      - catalog.analytics.product_performance
      - catalog.analytics.sales_summary

recipients:
  - name: acme_corp
    comment: ACME Corporation
    authentication_type: TOKEN
    email: data-team@acme.com

  - name: globex_inc
    comment: Globex Inc
    authentication_type: TOKEN
    email: analytics@globex.com

grants:
  - share: partner_analytics_share
    recipient: acme_corp
    permissions: [SELECT]

  - share: partner_analytics_share
    recipient: globex_inc
    permissions: [SELECT]

Best Practices

1. Security

• Use token-based authentication
• Implement row-level security when needed
• Regular audit of access permissions
• Monitor for unusual access patterns
• Rotate recipient tokens periodically

2. Performance

• Share aggregated data when possible
• Use partition pruning for large tables
• Monitor query performance
• Implement caching strategies
• Limit result set sizes

3. Governance

• Document all shares and recipients
• Maintain share configuration in version control
• Regular access reviews
• Clear data ownership
• Defined SLAs for shared data

4. Monitoring

# Set up alerts for anomalous usage
def check_usage_threshold(spark, share_name: str, threshold_gb: float = 100):
    """Alert if daily data transfer exceeds threshold."""
    query = f"""
    SELECT
        SUM(bytes_transferred) / 1024 / 1024 / 1024 as gb_transferred
    FROM system.access.audit
    WHERE share_name = '{share_name}'
      AND date_trunc('day', request_time) = current_date()
    """

    result = spark.sql(query).first()
    gb_transferred = result['gb_transferred'] or 0

    if gb_transferred > threshold_gb:
        send_alert(
            f"Share {share_name} transferred {gb_transferred:.2f}GB today, "
            f"exceeding threshold of {threshold_gb}GB"
        )

Common Pitfalls to Avoid

Don't:

• Share raw PII without review
• Neglect access auditing
• Skip documentation
• Ignore usage monitoring
• Hard-code credentials

Do:

• Implement data masking for sensitive fields
• Regular access audits
• Document all shares
• Monitor usage patterns
• Use secure credential management

Complete Examples

See /examples/ directory for:

• external_data_sharing.py: Complete external sharing setup
• multi_tenant_sharing.py: Multi-tenant implementation

Related Skills

• data-products: Share data products
• data-quality: Ensure shared data quality
• medallion-architecture: Share gold layer tables
• cicd-workflows: Automate share management

References

• Delta Sharing Protocol
• Databricks Delta Sharing Docs
• Delta Sharing Security