How to audit file accesses in DBFS on Databricks: step by step

This tutorial explains how to audit file accesses in DBFS on Databricks to record who read, wrote or deleted files. Auditing DBFS is useful for security, compliance and diagnosing issues related to shared data.

Prerequisites

• Databricks account with permissions to create notebooks and clusters.
• Cluster running with a runtime that supports Unity Catalog or workspace logs (recommended).
• Basic knowledge of PySpark/Python and filesystem commands on Databricks.

Step 1: Decide where to store the audit logs

It is important to choose a centralized location for the logs. You can use a container in Azure Blob / S3 or a folder in DBFS. I recommend a path in DBFS or a Delta table to simplify queries and retention.

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

Step 2: Implement a wrapper for DBFS operations

Create wrapper functions that perform the DBFS operation and then record an audit event with metadata (user, operation, path, timestamp, success/error). This avoids reliance on system logs and gives you full control.

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

Step 3: Wrapper for reading and removal with error capture

Similar to the previous step, implement functions to read and remove files. Also record size and hash when applicable to detect changes.

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

Step 4: Ingest events into a Delta table for analysis

To query and analyze the audit events, convert the NDJSON file to a Delta table. This makes aggregations, user searches and automatic retention easier.

# 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'")

Step 5: Useful queries and alerts

Create queries to identify suspicious accesses or frequent errors. You can use Databricks Workflows to schedule checks or integrate with alerts 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()

Verify the result

Confirm that events appear in the Delta table and that the data contains timestamps, user, operation and status. Perform write/read/delete operations with the wrappers and check for new rows in the table and in the NDJSON file.

1. Run write_file/read_file/remove_file and verify that audit_event records success/error.
2. Query audit_dbfs_events to see the most recent events.
3. Validate that the hashes and details match the expected content.

Conclusion

With a simple wrapper and ingestion to Delta it is possible to audit file accesses in DBFS, improving security and investigative capabilities. Next steps: automate log rotation, integrate with Unity Catalog/ACLs and create alerts in Databricks Workflows. Tip: start by auditing only a DBFS prefix to reduce noise and validate the event format.