Como auditar acessos a ficheiros em DBFS no Databricks: passo a passo

Este tutorial explica como auditar acessos a ficheiros em DBFS no Databricks para registar quem leu, escreveu ou eliminou ficheiros. Auditar o DBFS é útil para segurança, compliance e diagnóstico de problemas relacionados com dados partilhados.

Pré-requisitos

• Conta Databricks com permissões para criar notebooks e clusters.
• Cluster em execução com runtime que suporte o Unity Catalog ou os logs do workspace (recomendado).
• Conhecimentos básicos de PySpark/Python e comandos de sistema de ficheiros no Databricks.

Passo 1: Decidir onde guardar os logs de auditoria

É importante escolher um local centralizado para os logs. Pode usar um container em Azure Blob / S3 ou uma pasta em DBFS. Recomendo um caminho em DBFS ou uma tabela Delta para facilitar consultas e retenção.

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

Passo 2: Implementar wrapper para operações em DBFS

Crie funções wrapper que façam a operação no DBFS e, em seguida, registem um evento de auditoria com metadados (utilizador, operação, caminho, timestamp, sucesso/erro). Assim evita dependência dos logs do sistema e tem controlo 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

Passo 3: Wrapper para leitura e remoção com captura de erros

Semelhante ao passo anterior, implemente funções para ler e remover ficheiros. Registe também o tamanho e o hash quando aplicável para deteção de alterações.

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

Passo 4: Ingestão dos eventos para uma tabela Delta para análise

Para consultar e analisar os eventos de auditoria, converta o ficheiro NDJSON para uma tabela Delta. Isso facilita agregações, pesquisa por utilizador e retenção 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'")

Passo 5: Consultas úteis e alertas

Crie queries para identificar acessos suspeitos ou erros frequentes. Pode usar os Databricks Workflows para agendar verificações ou integrar com alertas via 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 os eventos aparecem na tabela Delta e que os dados contêm timestamps, user, operation e status. Faça operações de escrita/leitura/remover com os wrappers e verifique linhas novas na tabela e no ficheiro NDJSON.

1. Executar write_file/read_file/remove_file e verificar que audit_event regista sucesso/erro.
2. Consultar audit_dbfs_events para ver os eventos mais recentes.
3. Validar que os hashes e detalhes correspondem ao conteúdo esperado.

Conclusão

Com um wrapper simples e ingestão para Delta é possível auditar acessos a ficheiros em DBFS, melhorar a segurança e a capacidade de investigação. Próximos passos: automatizar a rotação de logs, integrar com o Unity Catalog/ACLs e criar alertas em Databricks Workflows. Dica: comece por auditar apenas um prefixo do DBFS para reduzir ruído e validar o formato de eventos.