Delta Sharing erklärt: Cross-Organisation-Datenaustausch ohne Datenkopien

Provider (Du)                         Recipient (externe Org)
---------------                       -----------------------
Unity Catalog                         Jeder kompatible Client
      |                                       |
      v                                       |
Delta Sharing Server  <----- REST API ----->  |
      |                                       |
      v                                       v
Cloud Storage  ------ Signed URLs -----> Direkter Read
(S3/ADLS/GCS)                            (keine Kopie)

-- Check if Delta Sharing is enabled
SHOW METASTORE;

-- Enable Delta Sharing on your metastore (admin required)
ALTER METASTORE SET ENABLE_DELTA_SHARING = TRUE;

-- Create a share
CREATE SHARE market_data_share
  COMMENT 'Live market data for external partners';

-- Add tables to the share
ALTER SHARE market_data_share
  ADD TABLE prod.gold.daily_prices;

-- Add a specific partition (e.g., only public data)
ALTER SHARE market_data_share
  ADD TABLE prod.gold.company_filings
  PARTITION (data_classification = 'public');

-- Add a table with an alias for the recipient
ALTER SHARE market_data_share
  ADD TABLE prod.silver.events
  AS prod_events_public
  COMMENT 'Anonymized event data';

-- Recipient with Unity Catalog (another Databricks workspace)
CREATE RECIPIENT acme_corp_recipient
  USING ID 'databricks.acme_corp_metastore_id';

-- Recipient without Unity Catalog (open sharing, uses tokens)
CREATE RECIPIENT external_partner_recipient
  COMMENT 'External analytics partner without Databricks';

-- Get the activation link for non-UC recipients
DESCRIBE RECIPIENT external_partner_recipient;
-- Returns an activation_link — send this to your recipient securely

-- Grant the share to the recipient
GRANT SELECT ON SHARE market_data_share TO RECIPIENT acme_corp_recipient;
GRANT SELECT ON SHARE market_data_share TO RECIPIENT external_partner_recipient;

-- Verify grants
SHOW GRANTS ON SHARE market_data_share;

-- Recipient: create a provider (references your metastore)
CREATE PROVIDER acme_data_provider
  USING ID 'databricks.your_metastore_share_id';

-- List available shares from the provider
SHOW SHARES IN PROVIDER acme_data_provider;

-- Create a catalog from the shared data
CREATE CATALOG shared_market_data
  USING SHARE acme_data_provider.market_data_share;

-- Query shared tables like any other table
SELECT * FROM shared_market_data.prod_gold.daily_prices
WHERE trade_date >= '2024-01-01'
LIMIT 100;

# pip install delta-sharing
import delta_sharing

# Load the profile file (downloaded from activation link)
profile_file = "path/to/config.share"

# List available tables
client = delta_sharing.SharingClient(profile_file)
shares = client.list_shares()
schemas = client.list_schemas(shares[0])
tables = client.list_tables(schemas[0])

print(f"Available tables: {[t.name for t in tables]}")

# Load as pandas DataFrame
df = delta_sharing.load_as_pandas(
    f"{profile_file}#market_data_share.prod_gold.daily_prices"
)
print(df.head())

# Load as Spark DataFrame (for large datasets)
spark_df = delta_sharing.load_as_spark(
    f"{profile_file}#market_data_share.prod_gold.daily_prices"
)
spark_df.filter("trade_date >= '2024-01-01'").show()

# Works with any Spark 3.x cluster
spark.conf.set("spark.sql.extensions", "io.delta.sharing.spark.DeltaSharingSparkExtension")

df = spark.read.format("deltaSharing") \
    .load("path/to/config.share#market_data_share.prod_gold.daily_prices")

df.filter("trade_date >= '2024-01-01'").show()

-- Share a specific table version (snapshot)
ALTER SHARE market_data_share
  ADD TABLE prod.gold.daily_prices
  AS daily_prices_snapshot_v100
  VERSION AS OF 100;

-- Share as of a specific timestamp
ALTER SHARE market_data_share
  ADD TABLE prod.gold.daily_prices
  AS daily_prices_jan_2024
  TIMESTAMP AS OF '2024-01-31T23:59:59Z';

# Recipient: read only new changes since last sync
df = delta_sharing.load_table_changes_as_pandas(
    f"{profile_file}#market_data_share.prod_gold.daily_prices",
    starting_version=150,
    ending_version=200
)
# Returns added/deleted rows with _change_type column
print(df[df["_change_type"] == "insert"])

SELECT
  event_time,
  event_type,
  source_ip_address,
  user_name,
  request_params.share_name,
  request_params.table_name
FROM system.access.audit
WHERE service_name = 'deltaSharingServer'
  AND event_date >= CURRENT_DATE - 30
ORDER BY event_time DESC;

-- Revoke a recipient's access to a share
REVOKE SELECT ON SHARE market_data_share FROM RECIPIENT external_partner_recipient;

-- Or drop the recipient entirely
DROP RECIPIENT external_partner_recipient;

ALTER SHARE market_data_share
  ADD TABLE prod.gold.company_profiles
  PARTITION (is_public = 'true');