Design A Database Batch Auditing Service - System Design

System requirements

The designed system enables users to define SQL, Hive, or Trino queries for periodic batch audits of their database tables. The system's capabilities and requirements include CRUD operations for audit jobs, specified intervals for audits, designated owners, and precise conditions based on SQL query results.

Functional requirements include:

A failed audit triggers an alert.
Users can view logs and the status of ongoing or past jobs.
Each job must complete within 6 hours, and each query within 15 minutes to prevent system delays.

Non-functional requirements cover:

Scalability, with less than 10,000 jobs anticipated.
Limited availability, as this is an internal tool without critical dependencies.
Security, ensuring that only job owners can modify their configurations.
Accuracy, ensuring that each audit's results reflect its configuration precisely.

This framework provides a robust structure for deploying a scalable and efficient SQL batch auditing service within an enterprise environment.

Capacity estimation

Estimate the scale of the system you are going to design...

API design

Define what APIs are expected from the system...

Database design

Defining the system data model early on will clarify how data will flow among different components of the system. Also you could draw an ER diagram using the diagramming tool to enhance your design...

High-level design

The high level diagram depicts the preliminary high-level architectural framework for a service designed to enable users to perform periodic validation checks on their data tables. This architecture assumes the use of a batch ETL service similar to Airflow. This service is responsible for storing Python scripts for batch jobs, executing these jobs as per scheduled intervals, maintaining the status and historical data of these jobs, and determining the truth value of audit conditions. Users interact with this system through a user interface (UI) that communicates with a backend to manage these processes:

Users interact with the batch ETL service through the UI to manage (create, read, update, delete) auditing jobs and to check their statuses and historical data.
Unlike the batch ETL service, the alerting functionalities are handled separately. This service does not possess API endpoints to initiate alerts or view alert histories directly. Instead, users access this information through the alerting service facilitated by the UI and backend.

Request flows

Process Flow for Creating a Batch Auditing Job:

When a user requests to create a batch auditing job, the backend service generates the validation.py script by inserting user-defined values into a predefined template. This lightweight script is held in memory on each backend service host.
The backend then forwards this script to the batch ETL service, which sets up the audit job and securely stores the script. Upon successful creation, it sends an HTTP 200 status code back to the backend service as confirmation.

Service Overview:

The batch auditing framework essentially acts as a front for the shared Batch ETL Service. It configures jobs based on parameters like ownership, scheduling (cron expressions), and specific database queries from supported types such as Hive, Trino, Spark, and SQL. The system maintains a primary SQL database that holds configurations for these audit jobs in a job_config table, with an associated owner table linking jobs to their respective owners.

Given the diverse nature of validation queries that support various SQL-like dialects, the batch ETL service interfaces with multiple shared database systems. In events of job failures or audit discrepancies, the service communicates with a centralized alerting system to notify concerned parties. Enhanced security is achieved through integration with an OpenID Connect service, ensuring authenticated interactions across the system.

Running and Monitoring Audit Jobs:

Audit jobs operate on a predefined schedule and consist of two main actions:

Execution of the database query.
Evaluation based on the query's results.

The creation of these jobs leverages a Python script within an Airflow service, facilitated by a script template that adapts to specific job requirements such as interval, query specifics, and conditional logic.

Detailed component design

Despite our best efforts to prevent invalid data from being stored, it is inevitable that some errors will occur. Auditing acts as an additional layer of validation checks.

A typical application for batch auditing is to verify large files, often larger than 1 GB, which originate outside our organization and over which we have no creation control. Single hosts processing and validating each row of such large files can be inefficient. Storing the data in a MySQL database and using the LOAD DATA command, which is more efficient than INSERT, followed by SELECT statements for auditing, provides a more rapid and manageable solution. If the data is stored on a distributed file system like HDFS, NoSQL solutions like Hive or Spark can be employed for fast parallel processing.

Even when audits reveal invalid data, there may be a decision to retain this "dirty" data rather than having no data at all. Batch auditing is particularly adept at identifying specific issues such as duplicate entries or missing data. Some validation processes, like anomaly detection algorithms, rely on previously ingested data to identify inconsistencies in newly ingested data.

Defining a Validation with a Conditional Statement on a SQL Query’s Result

A table consists of rows and columns, with any specific location within referred to as a cell, element, datapoint, or value. These terms are used interchangeably here.

Let's explore how manually defined validations are implemented using comparison operators on SQL query results. The outcome of an SQL query is treated as a two-dimensional array, termed "result," upon which we can apply conditional statements. We review examples of daily validations, analyzing only the data from the previous day. These validations, demonstrated through SQL queries, highlight the application of conditional statements on query results.

For instance, validations can be defined for:

Individual datapoints of a column: A simple validation might be ensuring that the "latest timestamp is less than 24 hours old".

SELECT COUNT(*) AS cnt FROM Transactions WHERE Date(timestamp) >= Curdate() - INTERVAL 1 DAY

Possible true conditional statements are result[0][0] > 0 and result['cnt'][0] > 0.

Multiple datapoints of a column: If a user should not make more than five purchases per day, a validation can alert if this limit is exceeded.

SELECT user_id, count(*) AS cnt FROM Transactions WHERE Date(timestamp) = Curdate() - INTERVAL 1 DAY GROUP BY user_id

The conditional statement would be result.length <= 5.

Multiple columns in a single row: Validating that sales using a specific coupon code do not exceed 100 per day.

SELECT count(*) AS cnt FROM Transactions WHERE Date(timestamp) = Curdate() - INTERVAL 1 DAY AND coupon_code = @coupon_code

The conditional statement would be result.length <= 100.

Multiple tables: For verifying that all sales in North America have appropriate country codes.

SELECT * FROM sales_na S JOIN country_codes C ON S.country_code = C.id WHERE C.region != 'NA';

The conditional statement would be result.length == 0.

Conditional statements on multiple queries: Raising an alert if the number of sales changes significantly from one week to the next.

SELECT COUNT() FROM sales WHERE Date(timestamp) = Curdate();
SELECT COUNT() FROM sales WHERE Date(timestamp) = Curdate() - INTERVAL 7 DAY;

The conditional statement could be Math.abs(result[0][0][0] – result[1][0][0]) / result[0][0][0] < 0.1.

These examples demonstrate the flexibility and necessity of incorporating manually defined validations in maintaining data integrity. While this section focused on SQL, these principles can be adapted to other query languages like HiveQL, Trino, or Spark, and even general-purpose programming languages for broader application in database management systems.

Simple SQL Batch Auditing Service

Let us delve into developing a straightforward script for auditing a SQL table and subsequently outline the steps to create a batch auditing job using this script.

An Audit Script

The fundamental structure of a batch auditing job involves the following steps:

Execute a database query.
Capture the query result into a variable.
Evaluate the variable against predefined conditions to determine compliance.

Here is a Python script example that performs a MySQL query to verify if the most recent timestamp in our transactions table is less than 24 hours old, and then outputs the result:

import mysql.connector
Establish database connection
cnx = mysql.connector.connect(user='admin', password='password', host='127.0.0.1', database='transactions')
cursor = cnx.cursor()
Define and execute the query
query = """
SELECT COUNT(*) AS cnt
FROM Transactions
WHERE Date(timestamp) >= Curdate() - INTERVAL 1 DAY
"""
cursor.execute(query)
results = cursor.fetchall()
Close connections
cursor.close()
cnx.close()
Check the condition and print the result
print(results[0][0] > 0)  # A simple condition to check the count

In scenarios requiring multiple database queries, the script can be expanded to handle various queries and compare their results, enhancing the robustness of the audit process.

An Audit Service

To scale this script into a full-fledged batch auditing service, we can generalize the script to enable user-specific configurations such as SQL databases, queries, and evaluation conditions. The approach involves creating a Python template file, named validation.py.template, which can dynamically insert user-defined parameters and execute within a batch processing framework like Airflow or a simple cron job.

Here’s a streamlined version of how such a service might be structured using Airflow:

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
import mysql.connector
import os
import pdpyras
Define the validation function
def validation():
    # Establish connection and execute queries
    results = []
    queries = [{'query': "..."}]  # Example placeholder for queries
    for query in queries:
        cnx = mysql.connector.connect(user='admin', password='password', host='127.0.0.1', database='transactions')
        cursor = cnx.cursor()
        cursor.execute(query['query'])
        results.append(cursor.fetchall())
        cursor.close()
        cnx.close()
    # Push results to Airflow XCom for cross-task communication
    ti.xcom_push(key='validation_result', value=results)
Define the alerting function
def alert():
    # Retrieve results from previous task
    result = ti.xcom_pull(key='validation_result')
    if result:
        # Trigger an alert if the audit fails
        routing_key = os.environ['PD_API_KEY']
        session = pdpyras.EventsAPISession(routing_key)
        session.trigger("Validation failed", "audit")
Set up the Airflow DAG
dag = DAG(
    'sql_audit_service',
    default_args={
        'depends_on_past': False,
        'email': ['[email protected]'],
        'email_on_failure': True,
        'email_on_retry': False,
        'retries': 1,
        'retry_delay': timedelta(minutes=5),
    },
    description='A DAG for SQL batch auditing',
    schedule_interval=timedelta(days=1),
    start_date=datetime(2023, 1, 1),
    catchup=False
)
t1 = PythonOperator(
    task_id='run_validation',
    python_callable=validation,
    dag=dag
)
t2 = PythonOperator(
    task_id='trigger_alert',
    python_callable=alert,
    dag=dag
)
t1 >> t2  # Define task dependency

Addressing Scalability and Alert Management:

The primary challenges in scaling involve the batch ETL and alerting services. Discussions on scalability focus on optimizing these components to handle an increasing load without performance degradation.

Audits are logged and results stored in SQL, with monitoring to track job performance and trigger alerts on failures. Decision-making around alert management considers whether the batch ETL service or the backend should handle alert initiation, with a leaning towards centralizing this responsibility within the backend to avoid mismanagement and reduce maintenance complexity.

Trade offs/Tech choices

Explain any trade offs you have made and why you made certain tech choices...

Failure scenarios/bottlenecks

Try to discuss as many failure scenarios/bottlenecks as possible.

Future improvements

To further refine our auditing capabilities, we can integrate features that allow dynamic adjustments to audits based on specific triggers, such as changes in data volume or detected anomalies. This proactive approach ensures our auditing processes remain adaptive and aligned with operational needs.

By implementing these strategies, we aim to maintain a high-performance, cost-efficient service that supports the dynamic needs of our users and organizational requirements.