タイトルで言いたいことは終わり、以上！

……というのも何なので少し実例を交えて。

使ったことがある方であればご存知の通り、ワークフローマネージャ Luigi ではあるタスクによって生成されるデータを抽象化して Target の継承クラスとして扱います。とりあえずこいつは exists() メソッドを実装してすでに存在しているかどうかが確認できれば機能します。
例えばローカルファイルであれば LocalTarget を使う訳ですが、とりあえずチュートリアルに従って書くと次のようになります。

from luigi import Task, ExternalTask, LocalTarget
import pandas as pd


class RawFile(ExternalTask):

    def output(self):
        return LocalTarget('path/to/file.csv')


class Aggregation(Task):

    def requires(self):
        yield RawFile()

    def run(self):
        df = pd.read_csv(self.input()[0], sep='\t', parse_dates=[7, 8], encoding='cp932')
        ...

さてさて……あ、ここで RawFile に依存する何か別のタスクが出てきてしまった。ということを考えましょう。徐にこんなコードを書き始める訳ですが……

class Plot(Task):

    def requires(self):
        yield RawFile()

    def run(self):
        df = pd.read_csv(

ちょっと待って。あなたは自分が関わっているすべてのプロジェクトについて、CSV ファイルのパーサに渡すべき適切なオプションをちゃんと覚えていますか？　私は覚えていません。もっと言えば毎回オプションを書きたいですか。私は断じて何回も書きたくありません。
せっかく Luigi が Target を元データのパスなんかを抽象化してくれているのに、これははっきり言ってダルいです。そうです、読み込みも抽象化してしまえば良いんです。

from luigi import Task, ExternalTask, LocalTarget
import pandas as pd


class RawFileTarget(LocalTarget):

    path = 'path/to/file.csv'

    def __init__(self):
        super(RawFileTarget, self).__init__(path)

    def load(self):
        return pd.read_csv(self.fn, sep='\t', parse_dates=[7, 8], encoding='cp932')

class RawFile(ExternalTask):

    def output(self):
        return RawFileTarget()

というように load() を定義した Target を定義してやりましょう。

そうすれば、どの依存タスクも

class Aggregation(Task):

    def requires(self):
        yield RawFile()

    def run(self):
        df = self.input()[0].load()
        ...


class Plot(Task):

    def requires(self):
        yield RawFile()

    def run(self):
        df = self.input()[0].load()
        ...

とスッキリしますね。後からデータハンドリングの担当者に「ごっめーん、以前渡されたアレ、ちょっと間違えちゃっててさ……」などということばとともにシレッと仕様の違う CSV ファイルを渡されようが、何を勘違いしたか渡されたファイルが Excel になっていようが、RawFileTarget.load() を書き換えれば済む話です。ハッピー！

タイトルに書いてあるとおり load() に対応するようにして save() を書いてやるとまたハッピーなのですが、コード例なんかは割愛します。