SOLID Principles for Data Pipelines

SOLID principle is five design rules for writing clean, maintainable, and scalable object-oriented computer code.

When I worked with Bukalapak, My team always encourage the SOLID principles when building product for our data platform. Therefore, I want to recall it so that I always remember to create a good software.

Single Responsibility Principle (SRP)

The official words are:

A class should have one reason to change.

My words are:

One class, one owner. If multiple teams request a change in a single class, we must divide the class into smaller ones.

Below is typical data pipeline script. It definitely violates the Single Responsibility principle. Details below:

class DailySalesJob:
    def fetch(self): pass
    def clean(self, rows): pass
    def to_csv(self, rows): pass
    def notify(self): pass

    def run(self):
        rows = self.fetch()
        rows = self.clean(rows)
        self.to_csv(rows)
        self.notify()

The script has 4 reason to change. They are:

The backend team want to change the API version.
The analyst team want to update the business logic.
The ops team want to change the output file to JSON format.
The data team want to change notification channel to Slack.

These actions need to be addressed by modifying the DailySalesJob class. As a result, this violates the Single Responsibility Principle.

Furthermore, We could face the following pains in the future:

Get a merge conflict, if multiple person edit the DailySalesJob class.
Re-test the Sales API, if the analyst team modify the business logic.

Resolution

Separate the DailySalesJob class into several classes, and ensure one class only has one owner.

Find the details below:

class SalesApiClient:
    def fetch(self): pass

class SalesCleaner:
    def clean(self, rows): pass

class CsvWriter:
    def write(self, rows): pass

class EmailNotifier:
    def notify(self, message): pass

class DailySalesJob:
    def __init__(self, client, cleaner, writer, notifier):
        self.client = client
        self.cleaner = cleaner
        self.writer = writer
        self.notifier = notifier

    def run(self):
        rows = self.client.fetch()
        rows = self.cleaner.clean(rows)
        self.writer.write(rows)
        self.notifier.notify("daily sales done")

As a result, When the analyst want to change the business logic, I only need to modify the SalesCleaner class.

Remarks

The DailySalesJob class still exists, and it's only for ordering the methods.

Open-Closed Principle (OCP)

The official words are:

Software should be open for extension, but closed for modification.

Open for extension means can add a new behavior.
Closed for modification means can add a new behavior without modify the old code that already works.

My words are:

New feature should be in a new class, not in if-else inside a working file.

The Exporter class below violates the Open/Closed principle because We need to modify the code that already running when adding a new switch for Avro format.

Find details below:

class Exporter:
    def write(self, rows, fmt, path):
        if fmt == "csv":
            # write csv
            ...
        elif fmt == "json":
            # write json
            ...
        elif fmt == "parquet":
            # write parquet
            ...
        else:
            raise ValueError(fmt)

Furthermore, We could face the following pains in the future:

Re-test others format when adding a new format.
Get merge conflict if multiple people modify the Exporter class.
Introduce a new switch inside a switch for compression, extension, etc.

Resolution

Separate the writer into several class and call it with abstraction.

Details below:

from abc import ABC, abstractmethod

class Writer(ABC):
    @abstractmethod
    def write(self, rows, path):
        pass

class CsvWriter(Writer):
    def write(self, rows, path):
        pass

class JsonWriter(Writer):
    def write(self, rows, path):
        pass

# add Parquet Writer class
# add Avro Writer class

class Exporter:
    def __init__(self, writer: Writer):
        self.writer = writer

    def run(self, rows, path):
        pass

WRITERS = {
    "json": JsonWriter,
    "csv": CsvWriter,
    # add parquet
    # add avro
    # add ocr
}    
exporter = Exporter(WRITERS["json"])

As a result, We don't need to modify the code that already running when adding a new format.

Remarks Do not implement this principle when there is no use case to write into multiple format. Instead, limit the Exporter class to only support one format. Additionally, We could implement this if there is a need to support to another format.

Liskov Substitution Principle (LSP)

The official words are:

If S is a subtype of T, then objects of type T can be replaced with objects of type S without breaking the program.

My words are:

Child classes must be able to replace parent classes without breaking the program.

The script appears to be working correctly until the caller writes the rows into the S3 bucket. The violation in this script is that the Exporter class breaks the promise made to the contract. Specifically, in the contract for the path value, for instance, path/to/dir.

Details below:

from abc import ABC, abstractmethod

class Writer(ABC):
    @abstractmethod
    def write(self, rows, path):
        pass

class S3Writer(Writer):
    def write(self, rows, path):
        if not path.startswith("s3://"):
            raise ValueError("path must be starts with s3://")
        pass

class JsonWriter(Writer):
    def write(self, rows, path):
        pass

class Exporter:
    def __init__(self, writer: Writer):
        self.writer = writer

    def run(self, rows, path):
        self.writer.write(rows, path)

rows = "id, name, status"
path = "path/to/dir"
Exporter(writer=S3Writer).run(rows=rows, path=path)

If you use the following code, you haven’t implemented Liskov Substitution. Furthermore, adding validation for the GCS path will violate the Open/Close principle.

class Exporter:
    def __init__(self, writer: Writer):
        self.writer = writer

    def run(self, rows, path):
        if isinstance(self.writer, S3Writer):
            self.writer.write(rows, f"s3://{path}")
        else:
            self.writer.write(rows, path)

As a result, substitute the switch with creating an attributes in the S3Writer class constructor.

Find details below:

from abc import ABC, abstractmethod

class Writer(ABC):
    @abstractmethod
    def write(self, rows, path):
        pass

class S3Writer(Writer):
    def __init__(self, bucket):
        self.bucket = bucket

    def write(self, rows, path):
        path = f"{self.bucket}/{path}"
        if not path.startswith("s3://"):
            raise ValueError("path must be starts with s3://")

class JsonWriter(Writer):
    def write(self, rows, path):
        pass

class Exporter:
    def __init__(self, writer: Writer):
        self.writer = writer

    def run(self, rows, path):
        self.writer.write(rows, path)

rows = "id, name, status"
path = "path/to/dir"
s3_bucket = "s3://bucket_name"

Exporter(writer=S3Writer(s3_bucket)).run(rows=rows, path=path)

Interface Segregation Principle (ISP)

The official words are:

Client should not be forced depends on methods they don't used. The clients here is the code that use the Interface.

My words are:

A class should not be forced to use methods or interfaces it does not need.

Example below:

from abc import ABC, abstractmethod

def DataStore(ABC):
    @abstractmethod
    def read(): pass

    @abstractmethod
    def write(): pass

    @abstractmethod
    def list_partitions(): pass

    @abstractmethod
    def vacuum(): pass

def CsvStore(DataStore):
    def read(): pass
    def write(): pass
    def list_partitions(): raise NotImplementedError("The list_partitions method is not implemented.")
    def vacuum(): raise NotImplementedError("The vacuum method is not implemented")

def IcebergStore(DataStore):
    def read(): pass
    def write(): pass
    def list_partitions(): pass
    def vacuum(): pass

The fix is that I break the "Big" Interface to a few Interfaces contains relevant methods. Details below:

from abc import ABC, abstractmethod

class Readable(ABC):
    @abstractmethod
    def read(self): pass

class Writeable(ABC):
    @abstractmethod
    def write(self): pass

class Partitionable(ABC):
    @abstractmethod
    def list_partitions(self): pass

    @abstractmethod
    def vacuum(self): pass

class CsvStore(Readable, Writeable):
    def write(self): pass
    def read(self): pass

class IcebergStore(Readable, Writeable, Partitionable):
    def write(self): pass
    def read(self): pass
    def list_partitions(self): pass
    def vacuum(self): pass

Dependency Inversion Principle (DIP)

The official words are:

The high-level module should not depend on the low-level module. Both should depend on abstractions.

my words are:

The important logic should not be glued with specific tool. Both should be communicated within contract.

Details below:

High-level: The business logic. For example: a class contains logic for sales report.
Low-level: The tool. For example: Postgres, Kafka, GCS, etc.

An example below:

class PostgresLoader:
    def load(self, rows):
        pass

class SalesReport:
    def __init__(self):
        self.loader = PostgresLoader()

    def run(self, rows):
        self.loader.load(rows)

The script is definitely violates the principle because the SalesReport class (High-level) depend on a tools which is Postgres (Low-level).

The pain feel later are:

We need a Postgres to unit-test the script.
We need to change the SalesReport class if We want to change the storage destination to BigQuery.

Find the fix below:

from abc import ABC, abstractmethod

class Loader(ABC):
    @abstractmethod
    def load(self):
        pass

class PostgresLoader(Loader):
    def load(self):
        pass

class BigQueryLoader(Loader):
    def load(self):
        pass

class SalesReport:
    def __init__(self, loader: Loader):
        self.loader = loader

    def run(self):
        self.loader.load()


SalesReport(PostgresLoader).run()   # Load to Postgres
SalesReport(BigQueryLoader).run()   # Load to BigQuery