Contesto

Per i clienti che dispongono di strumenti di visualizzazione dei dati che non funzionano con il metodo di condivisione dei dati da databricks, utilizzeremo questa implementazione Python per supportare un inserimento dati efficace.

Introduzione

Delta Sharing è un protocollo aperto sviluppato da Databricks per la condivisione sicura dei dati in tempo reale tra organizzazioni. Consente a un fornitore di dati di condividere i set di dati con i destinatari senza richiedere loro di utilizzare Databricks o qualsiasi piattaforma specifica.

Questa guida ti spiega come consumare i dati condivisi tramite Delta Sharing e inserirli nel tuo database. Forniamo uno script di riferimento Python pronto all'uso che legge una tabella Delta condivisa e la scrive in un database PostgreSQL.

Per chi è pensato?

Questa guida è per i clienti che ricevono dati attraverso Delta Sharing e desiderano caricarli in un sistema che non supporta nativamente il formato Delta — come PostgreSQL, MySQL, o qualsiasi database SQL.

Panoramica dell'architettura

Il flusso di acquisizione è semplice:

Non è necessario un account Databricks per il destinatario.

Nessun driver speciale — lo script utilizza la libreria Python open-source delta-sharing.

I dati rimangono sotto il tuo controllo — una volta acquisiti, i dati risiedono nel tuo database.

Prerequisiti

Prima di iniziare, assicurati di avere quanto segue:

Passaggio 1 — Ottieni il tuo profilo Delta Sharing

Il tuo fornitore di dati (Uberall) ti fornirà un file di profilo Delta Sharing. Si tratta di un piccolo file JSON con estensione .share che contiene i dettagli di connessione e il token di autenticazione.

Il file è simile a questo:

{
  "shareCredentialsVersion": 1,
  "endpoint": "https://<databricks-host>/api/2.0/delta-sharing/",
  "bearerToken": "<your-token>"
}

Salva questo file in una posizione sicura sulla tua macchina. Farai riferimento al suo percorso durante la configurazione dello script.

Importante:

Questo file contiene un bearer token che consente l'accesso ai dati condivisi. Trattalo come una password.

Non inserirlo nel controllo della versione.

Limita le autorizzazioni del file in modo che solo il tuo utente possa leggerlo:

chmod 600 /path/to/your/profile.share

Se non hai ricevuto il tuo file di profilo, contatta il team Uberall Data Engineering.

Passaggio 2 — Clona il Repository

Apri un terminale e clona il repository:

git clone https://gitlab.com/momentfeed/uberall/development/ar/data-lake/lakehouse/databricks/delta-sharing-code-examples.git

Accedi alla directory del progetto:

cd delta-sharing-code-examples

Il repository contiene i seguenti file:

Passaggio 3 – Configurare un ambiente Python

Ti consigliamo di usare un ambiente virtuale per evitare conflitti con altri progetti Python.

Crea e attiva un ambiente virtuale:

# Create the virtual environment
python -m venv venv

# Activate it (Linux / macOS)
source venv/bin/activate

# Activate it (Windows PowerShell)
.\venv\Scripts\Activate.ps1

Installa le dipendenze richieste:

pip install -r requirements.txt

Questo installa i seguenti pacchetti:

Passaggio 4 – Configura un database PostgreSQL

Hai due opzioni:

Opzione A: Usa la configurazione Docker inclusa (consigliata per i test)

Se hai Docker installato, il repository include un file compose.yaml che avvia un'istanza PostgreSQL 16 con un singolo comando.

Avvia il database:

docker compose up -d

Verifica che sia in esecuzione:

docker compose logs db

Cerca database system is ready to accept connections nell'output.

Test di connettività rapido:

docker compose exec db psql -U postgres -d testdb -c "SELECT 1;"

L'istanza Docker di PostgreSQL utilizza i seguenti valori predefiniti:

Questi corrispondono alla configurazione predefinita nello script, quindi non sono necessarie modifiche aggiuntive.

Quando hai finito i test, arresta il database con:

# Stop containers (data is preserved)
docker compose down

# Stop and delete all data
docker compose down -v

Opzione B: Usa la tua istanza PostgreSQL

Se disponi già di un database PostgreSQL, annota semplicemente i dettagli di connessione (host, porta, database, utente, password). Li inserrai nel Passaggio 5.

Passaggio 5 — Configura lo script

Apri delta_sharing_to_postgres.py in un editor di testo qualsiasi. All'inizio del file troverai la sezione USER CONFIGURATION:

# =============================================================================
# USER CONFIGURATION — Edit the values below to match your environment.
# =============================================================================

DELTA_SHARING_PROFILE_PATH = "/path/to/your/profile.share"
DELTA_SHARING_TABLE = "<share_name>.<schema_name>.<table_name>"

PG_HOST = "localhost"
PG_PORT = 5432
PG_DATABASE = "testdb"
PG_USER = "postgres"
PG_PASSWORD = "example"

PG_TARGET_SCHEMA = "public"
PG_TARGET_TABLE = "delta_sharing_data"

WRITE_MODE = "replace"

Modifica i seguenti valori:

Come scoprire le tabelle disponibili

Se non conosci il nome esatto della tabella, puoi eseguire uno script di individuazione rapida. Crea un file Python temporaneo o usa una shell Python:

import delta_sharing

client = delta_sharing.SharingClient("/path/to/your/profile.share")

for table in client.list_all_tables():
    print(f"{table.share}.{table.schema}.{table.name}")

In questo modo vengono visualizzate tutte le tabelle a tua disposizione. Usa la stringa <share>.<schema>.<table> completa come valore per DELTA_SHARING_TABLE.

In alternativa, esegui semplicemente lo script principale – registra tutte le tabelle disponibili all'inizio prima di tentare di leggere i dati.

Modalità di scrittura

L'impostazione WRITE_MODE controlla il modo in cui lo script gestisce i dati esistenti nella tabella di destinazione:

Passaggio 6 — Esegui lo Script

Assicurati che il tuo ambiente virtuale sia attivato (vedi Passaggio 3), quindi esegui:

Cosa Aspettarsi

Un'esecuzione riuscita produce un output simile al seguente:

2026-02-20 17:20:00,000 [INFO] ============================================================
2026-02-20 17:20:00,000 [INFO] Delta Sharing to PostgreSQL — Ingestion Script
2026-02-20 17:20:00,000 [INFO] ============================================================
2026-02-20 17:20:00,001 [INFO] Loading Delta Sharing profile from: ./config.share
2026-02-20 17:20:00,002 [INFO] Available shares and tables:
2026-02-20 17:20:01,500 [INFO]   my_share.analytics.sales_data
2026-02-20 17:20:01,500 [INFO] Reading table: my_share.analytics.sales_data
2026-02-20 17:20:05,000 [INFO] Read 15000 rows and 12 columns.
2026-02-20 17:20:05,001 [INFO] Preview of the data (first 5 rows):
2026-02-20 17:20:05,002 [INFO]
   col_a  col_b  col_c ...
0  ...    ...    ...
1  ...    ...    ...
2026-02-20 17:20:05,100 [INFO] Connecting to PostgreSQL...
2026-02-20 17:20:05,200 [INFO] PostgreSQL connection successful.
2026-02-20 17:20:05,201 [INFO] Writing 15000 rows to public.delta_sharing_data (mode=replace)...
2026-02-20 17:20:08,000 [INFO] Write complete.
2026-02-20 17:20:08,100 [INFO] Validation: public.delta_sharing_data now contains 15000 rows.
2026-02-20 17:20:08,101 [INFO] ============================================================
2026-02-20 17:20:08,101 [INFO] Ingestion completed successfully.
2026-02-20 17:20:08,101 [INFO] ============================================================

Lo script:

Si connette a Delta Sharing e elenca tutte le tabelle disponibili.

Legge la tabella specificata in memoria.

Mostra un'anteprima delle prime 5 righe.

Scrive i dati nella tua istanza PostgreSQL.

Convalida l'ingestion contando le righe nella tabella di destinazione.

Passaggio 7 — Verifica i dati

Dopo un'esecuzione riuscita, puoi verificare i dati in PostgreSQL.

Usando la riga di comando (setup Docker):

docker compose exec db psql -U postgres -d testdb -c "SELECT COUNT(*) FROM public.delta_sharing_data;"

Usando un client SQL (DataGrip, DBeaver, pgAdmin, ecc.):

Connettiti alla tua istanza PostgreSQL con le stesse credenziali del Passaggio 5, quindi esegui:

-- Row count
SELECT COUNT(*) FROM public.delta_sharing_data;

-- Preview rows
SELECT * FROM public.delta_sharing_data LIMIT 10;

-- Check column types
SELECT column_name, data_type
FROM information_schema.columns
WHERE table_schema = 'public' AND table_name = 'delta_sharing_data';

Casi d'uso

Aggiornamento completo dello snapshot

Imposta WRITE_MODE = "replace" ed esegui lo script secondo una pianificazione (ad es., giornalmente tramite un cron job). Ogni esecuzione sostituisce l'intera tabella con i dati più recenti dalla condivisione.

# Example cron entry: run every day at 2:00 AM
0 2 * * * /path/to/venv/bin/python /path/to/delta_sharing_to_postgres.py

Aggiungi / accumula dati nel tempo

Imposta WRITE_MODE = "append" per continuare ad aggiungere righe a ogni esecuzione. Questo è utile quando la tabella condivisa contiene nuovi record (ad es., eventi giornalieri o log) e vuoi creare una cronologia nel tuo database.

Nota: Questa modalità non esegue la deduplicazione. Se esegui lo script due volte con gli stessi dati, avrai righe duplicate. Valuta l'aggiunta di logica di deduplicazione nel tuo database o nello script se questo è un problema.

Carica in un database diverso

Lo script utilizza SQLAlchemy, che supporta molti database. Per utilizzare un database diverso, modifica la stringa di connessione in build_pg_connection_string() e installa il driver appropriato:

Incorpora in un data warehouse o database cloud

Lo stesso approccio funziona con database ospitati su cloud (Amazon RDS, Google Cloud SQL, Azure Database for PostgreSQL, ecc.). Aggiorna semplicemente i parametri di connessione per puntare alla tua istanza cloud.

Guida alla personalizzazione

Lo script è intenzionalmente semplice e pensato per essere adattato alle tue esigenze. Di seguito troverai le modifiche più comuni.

Filtrare o trasformare i dati prima del caricamento

Dopo aver letto i dati, puoi manipolare il DataFrame di Pandas prima di scriverlo nel database. Ad esempio, per mantenere solo colonne specifiche:

df = read_delta_sharing_table(DELTA_SHARING_PROFILE_PATH, DELTA_SHARING_TABLE)

# Keep only the columns you need
df = df[["column_a", "column_b", "column_c"]]

# Rename columns to match your schema
df = df.rename(columns={"column_a": "id", "column_b": "name"})

# Filter rows
df = df[df["status"] == "active"]

ingest_to_postgres(df, connection_string, PG_TARGET_SCHEMA, PG_TARGET_TABLE, WRITE_MODE)

Caricamento incrementale (carica solo i nuovi dati)

Se la tabella condivisa dispone di una colonna di timestamp o data, puoi filtrare per caricare solo i nuovi record:

df = read_delta_sharing_table(DELTA_SHARING_PROFILE_PATH, DELTA_SHARING_TABLE)

# Only load data from the last 7 days
from datetime import datetime, timedelta
cutoff = datetime.now() - timedelta(days=7)
df = df[df["updated_at"] >= cutoff]

ingest_to_postgres(df, connection_string, PG_TARGET_SCHEMA, PG_TARGET_TABLE, "append")

Carica più tabelle

Duplica il blocco di configurazione o scorri un elenco di tabelle:

TABLES_TO_INGEST = [
    {"delta_table": "my_share.schema.table_a", "pg_table": "table_a"},
    {"delta_table": "my_share.schema.table_b", "pg_table": "table_b"},
]

for entry in TABLES_TO_INGEST:
    df = read_delta_sharing_table(DELTA_SHARING_PROFILE_PATH, entry["delta_table"])
    ingest_to_postgres(df, connection_string, PG_TARGET_SCHEMA, entry["pg_table"], WRITE_MODE)

Usa variabili di ambiente al posto di valori hardcoded

Per un uso in produzione, evita di codificare i segreti nello script. Usa invece le variabili di ambiente:

import os

DELTA_SHARING_PROFILE_PATH = os.environ["DELTA_SHARING_PROFILE_PATH"]
PG_HOST = os.environ.get("PG_HOST", "localhost")
PG_PASSWORD = os.environ["PG_PASSWORD"]

Quindi impostali prima di eseguire:

export DELTA_SHARING_PROFILE_PATH="./config.share"
export PG_PASSWORD="your_password"
python delta_sharing_to_postgres.py

Procedure consigliate per la sicurezza

Risoluzione dei problemi

Assistenza

Se riscontri problemi relativi a:

I dati condivisi, i token di accesso o i permessi — contatta il team Uberall Data Engineering.

Lo script di acquisizione — consulta questa guida e il README.md nel repository. Lo script è fornito come implementazione di riferimento; sei responsabile dell'adattamento e della manutenzione per il tuo ambiente.