Cómo auditar accesos a archivos en DBFS en Databricks: paso a paso

Este tutorial explica cómo auditar accesos a archivos en DBFS en Databricks para registrar quién leyó, escribió o eliminó archivos. Auditar el DBFS es útil para seguridad, cumplimiento y diagnóstico de problemas relacionados con datos compartidos.

Prerequisitos

• Cuenta Databricks con permisos para crear notebooks y clusters.
• Cluster en ejecución con runtime que soporte el Unity Catalog o los logs del workspace (recomendado).
• Conocimientos básicos de PySpark/Python y comandos de sistema de archivos en Databricks.

Paso 1: Decidir dónde guardar los logs de auditoría

Es importante elegir un lugar centralizado para los logs. Puede usar un container en Azure Blob / S3 o una carpeta en DBFS. Recomiendo una ruta en DBFS o una tabla Delta para facilitar consultas y retención.

# exemplo: caminho em DBFS
audit_path = '/dbfs/audit/logs/dbfs_access_audit/'

Paso 2: Implementar wrapper para operaciones en DBFS

Crear funciones wrapper que realicen la operación en DBFS y, a continuación, registren un evento de auditoría con metadatos (usuario, operación, ruta, timestamp, éxito/error). Así se evita dependencia de los logs del sistema y se tiene control total.

import os
import json
from datetime import datetime
from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

AUDIT_PATH = '/dbfs/audit/logs/dbfs_access_audit/events.ndjson'

def write_audit(event):
    event_line = json.dumps(event, default=str)
    # escreve em append num ficheiro NDJSON em DBFS
    with open(AUDIT_PATH.replace('/dbfs',''), 'a') as f:
        f.write(event_line + '\n')

def audit_event(user, operation, path, status, details=None):
    event = {
        'timestamp': datetime.utcnow().isoformat() + 'Z',
        'user': user,
        'operation': operation,
        'path': path,
        'status': status,
        'details': details
    }
    write_audit(event)

# exemplo de wrapper para escrever ficheiro
from pathlib import Path

def write_file(user, path, content):
    try:
        full_path = path.replace('dbfs:','/dbfs')
        parent = os.path.dirname(full_path)
        os.makedirs(parent, exist_ok=True)
        with open(full_path, 'w') as f:
            f.write(content)
        audit_event(user, 'write', path, 'success')
    except Exception as e:
        audit_event(user, 'write', path, 'error', str(e))
        raise

Paso 3: Wrapper para lectura y eliminación con captura de errores

Similar al paso anterior, implemente funciones para leer y eliminar archivos. Registre también el tamaño y el hash cuando sea aplicable para detección de cambios.

import hashlib

def file_hash(path):
    full_path = path.replace('dbfs:','/dbfs')
    h = hashlib.sha256()
    with open(full_path, 'rb') as f:
        for chunk in iter(lambda: f.read(8192), b''):
            h.update(chunk)
    return h.hexdigest()

def read_file(user, path):
    try:
        full_path = path.replace('dbfs:','/dbfs')
        with open(full_path, 'r') as f:
            content = f.read()
        h = file_hash(path)
        audit_event(user, 'read', path, 'success', {'sha256': h})
        return content
    except Exception as e:
        audit_event(user, 'read', path, 'error', str(e))
        raise

def remove_file(user, path):
    try:
        full_path = path.replace('dbfs:','/dbfs')
        os.remove(full_path)
        audit_event(user, 'delete', path, 'success')
    except Exception as e:
        audit_event(user, 'delete', path, 'error', str(e))
        raise

Paso 4: Ingesta de los eventos a una tabla Delta para análisis

Para consultar y analizar los eventos de auditoría, convierta el archivo NDJSON a una tabla Delta. Esto facilita agregaciones, búsqueda por usuario y retención automática.

# ler NDJSON e gravar Delta
events_df = spark.read.json('dbfs:/audit/logs/dbfs_access_audit/events.ndjson')
events_df.write.format('delta').mode('overwrite').save('/mnt/delta/audit_dbfs_events')

# criar tabela gerida ou externa (opcional)
spark.sql("CREATE TABLE IF NOT EXISTS audit_dbfs_events USING DELTA LOCATION '/mnt/delta/audit_dbfs_events'")

Paso 5: Consultas útiles y alertas

Crear consultas para identificar accesos sospechosos o errores frecuentes. Puede usar los Databricks Workflows para programar verificaciones o integrar con alertas vía webhook.

# exemplo de query: top utilizadores com erros
spark.sql("SELECT user, count(*) as errors FROM audit_dbfs_events WHERE status='error' GROUP BY user ORDER BY errors DESC LIMIT 10").show()

# exemplo: ficheiros mais lidos
spark.sql("SELECT details.sha256 as hash, count(*) as reads FROM audit_dbfs_events WHERE operation='read' AND status='success' GROUP BY details.sha256 ORDER BY reads DESC LIMIT 10").show()

Verificar o resultado

Confirme que los eventos aparecen en la tabla Delta y que los datos contienen timestamps, user, operation y status. Haga operaciones de escritura/lectura/eliminación con los wrappers y verifique filas nuevas en la tabla y en el archivo NDJSON.

1. Ejecutar write_file/read_file/remove_file y verificar que audit_event registra éxito/error.
2. Consultar audit_dbfs_events para ver los eventos más recientes.
3. Validar que los hashes y detalles corresponden al contenido esperado.

Conclusión

Con un wrapper simple y la ingesta a Delta es posible auditar accesos a archivos en DBFS, mejorar la seguridad y la capacidad de investigación. Próximos pasos: automatizar la rotación de logs, integrar con el Unity Catalog/ACLs y crear alertas en Databricks Workflows. Consejo: empiece por auditar solo un prefijo del DBFS para reducir el ruido y validar el formato de eventos.