cover_effective_python

【 Effective Python, 2nd Edition 】@classmethod ポリモーフィズム ( polymorphism ) を利用して、複数の派生クラスをよりジェネリック ( generic ) に活用しよう! 投稿一覧へ戻る

Tags: Python , Effective , polymorphism , generic , classmethod

Published 2020年7月8日23:11 by T.Tsuyoshi

Python では、インスタンスに限らずクラスでもポリモーフィズム ( polymorphism ) を活用できます。
これはどういうことなんでしょうか? そして、どんなときに活用できるんでしょうか?


Polymorphism は、クラス階層内におけるそれぞれの派生クラスが、同じインターフェースながら機能的には異なる独自のものを提供することを可能にします。


今回目指すのは MapReduce モデルの実装です。
MapReduce モデルは、ある処理を複数のサブプロセスに分割して実行し(map)、それぞれの結果を統合して1つの結論を導く(reduce)、分散コンピューティングモデルの1つです。


まずはデータを提供するためのベースとなる抽象クラスを用意します。
このクラスの派生クラスでは read() という共通インターフェースを提供します。


class InputData:
"""データを提供するクラスのベース抽象クラス"""
def read(self):
raise NotImplementedError


class PathInputData(InputData):
"""データをファイルから読み込んで提供する実装クラス"""
def __init__(self, path):
super().__init__()
self.path = path

def read(self):
with open(self.path) as f:
return f.read()



取得したデータを MapReduce モデルで処理するためのクラスも同様に用意します。


class Worker:
""" MapReduce モデルクラスのベース抽象クラス"""
def __init__(self, input_data: InputData):
self.input_data = input_data
self.result = None

def map(self):
raise NotImplementedError

def reduce(self, other):
raise NotImplementedError


class LineCountWorker(Worker):
"""
このプログラムにおける MapReduce モデル実装クラス。
取得したデータ内に含まれる '\n' 文字の数をカウントする。
"""
def map(self):
data = self.input_data.read()
self.result = data.count('\n')

def reduce(self, other: Worker):
self.result += other.result



なかなかの実装になりました、満足満足!


役割ごとにベースクラスを設け、それを抽象クラス ( abstraction class ) といたしますことで、 OOP (オブジェクト指向プログラミング; Object Oriented Programming ) におけるポリモーフィズム ( polymorphism ) を具現し、そこから派生したクラスにおける一貫したインターフェース提供を可能といたしました!


新商品のプレゼンのような文言になりましたが...


が、プログラムですから、オブジェクトを作成し、それらが連携して初めて目標とするところの MapReduce モデルを実現できるわけです。


では、これらの部品部品をどうつなぎ合わせればいいのでしょう?
誰がオブジェクトを作り、どうやって各オブジェクトを制御して MapReduce を実現すればいいのでしょう?


真っ先に思い浮かぶ方法は、複数のヘルパー関数を用意し、オブジェクトを1つ1つ手作業で作成、結び付ける方法です。


まず手始めに、指定されたディレクトリ内のファイル1つ1つに対して PathInputData オブジェクトを作成します。


import os
from typing import Generator, List


def generate_inputs(data_dir) -> Generator:
for file in os.listdir(data_dir):
yield PathInputData(os.path.join(data_dir, file))



続いて、generate_inputs ジェネレータによって生成される InputData インスタンスを渡して LineCountWorker インスタンスを作成します。


def create_workers(input_list: Generator) -> List[Worker]:
workers = []
for input_data in input_list:
workers.append(LineCountWorker(input_data))
return workers



ここで作成した Worker インスタンスの map メソッドを複数のスレッドで実行し ( ここが MapReduce モデルにおける Map ステージになります )、続けて 1つの Worker インスタンスの reduce メソッドを繰り返し呼び出してそれぞれの map メソッドの結果を集計し最終的な値を取得します ( ここが Reduce ステージです )。


from threading import Thread


def execute(workers) -> int:
threads = [Thread(target=w.map) for w in workers]
for thread in threads:
thread.start()
for thread in threads:
thread.join()

first, *rest = workers
for worker in rest:
first.reduce(worker)
return first.result



全てのヘルパー関数が完成しました。
最後に、これらの関数を順次実行していくための関数を作りましょう。


def mapreduce(data_dir) -> int:
inputs = generate_inputs(data_dir)
workers = create_workers(inputs)
return execute(workers)



ではここで、動作テストに使用するファイルを作成しちゃいましょう。
1つ1つのファイルには改行文字 ( '\n' ) をランダムな個数書き込んでいるだけです。


import random


def prepare_test_files(tempdir):
os.makedirs(tempdir)
for i in range(100):
with open(os.path.join(tempdir, str(i)), 'w') as f:
f.write('\n' * random.randint(0, 100))


tempdir = 'test_files'
prepare_test_files(tempdir)



さあ、実行しましょう!


result = mapreduce(tempdir)

print(f"全てのファイルの行数は合計 {result} 行です。")
# 全てのファイルの行数は合計 4825 行です。



さて、この MapReduce モデルの実装方法における問題点は何でしょう?


それは、何はともあれ mapreduce() 関数がまるっきりジェネリック ( generic ) ではない、ということです。


もし、InputData クラス、Worker クラスのまったく新しいサブクラスを作成した場合、それらを利用する generate_inputs()、create_workers()、mapreduce() の各関数をすべて書き換えなければなりません。


では問題の根源はヘルパー関数の作り方にあるのでしょうか?
違います。実は InputData クラス、Worker クラス自体がすでにジェネリックではないんです。


例えば InputData クラスを元に複数の派生クラスを作成した場合、それらを一貫して呼び出す方法が存在していないんです。
ですから利用する側ではそれぞれの派生クラスのインスタンスをその都度作成して、別々に呼び出さなければいけなくなっているんです。


ではどうしたらいいんでしょう?
同じクラスをベースとするサブクラスを一貫した方法で呼び出す、それにはコンストラクタポリモーフィズム ( constructor polymorphism ) を実装すればいいんです。


が、残念なお知らせです。
Python で許されているクラスコンストラクターはただ一つ、__init__() だけですよね。。
ただ一種類の __init__() で複数のサブクラスを区別して呼び出す? いえいえ、それはさすがに無理です。結局それぞれのサブクラスを別々にインスタンス化することと何ら変わりません。


ではどうすれば? さてどうしましょう?
解決方法は、クラスメソッドポリモーフィズム ( class method polymorphism ) の実装です。


これは InputData.read() を各サブクラスで再定義してそれぞれに異なる機能を持たせているのと基本的に同じですが、これが各サブクラスのインスタンス ( オブジェクト ) に作用する一方、クラスメソッドポリモーフィズムはクラスそのもの自体に作用する点で大きく異なります。


つまり、ベースクラスでクラスメソッドの雛形を作っておき、必要に応じてサブクラスで機能の上書きを行います。
あとは、呼び出し元から利用するクラス名だけを指定すれば、[サブクラス名.クラスメソッド名] の構文でベースクラスが共通する全てのサブクラスを一貫して呼び出すことができるインターフェースを提供できるようになります。


ここまで説明したコンセプトを元に InputData クラスを書き換えます。
このクラスをベースクラスとする全てのサブクラスを一貫して呼び出せるようにクラスメソッドの雛形を追加します。


class GenericInputData:
"""
データを提供するクラスのベース抽象クラス
全てのサブクラスは、[サブクラス名.generate_inputs(config)] という共通インターフェースでインスタンス化でき、
read() でデータの取出しが行えます。
"""
def read(self):
raise NotImplementedError

@classmethod
def generate_inputs(cls, config: dict):
raise NotImplementedError



ここでちょっとしたトリックを使っています。
各サブクラスのデータの取得先はファイルかもしれないし、インターネットかもしれません。それぞれ異なるはずです。
そうした取得先データを一貫して渡すために、必要な情報をセットした辞書を渡すようにしています。
これによってサブクラスごとにインターフェースが異なる、ということがなくなります。


GenericPathInputData サブクラスではこの config パラメータからファイルを読み込むためのディレクトリ名を取得しています。


class GenericPathInputData(GenericInputData):
"""データをファイルから読み込んで提供する実装クラス"""
def __init__(self, path):
super().__init__()
self.path = path

def read(self):
with open(self.path) as f:
return f.read()

@classmethod
def generate_inputs(cls, config: dict) -> Generator:
data_dir = config['data_dir']
for file in os.listdir(data_dir):
yield cls(os.path.join(data_dir, file))



Worker クラスについても、派生クラスに共通するインターフェースを提供するためにクラスメソッドを追加しています。


このクラスメソッドでは、データ提供元として GenericInputData クラスの派生クラスを受け取るようにしていますが、
今や GenericInputData クラスは generate_inputs クラスメソッドを実装していますので、[クラス名.generate_inputs(config)] とすることでサブクラスの違いを意識することなくクラスインスタンスを取得できるようになっています。


class GenericWorker:
"""
MapReduce モデルクラスのベースクラス。
create_workers クラスメソッドは各サブクラスに共通した機能、インターフェースを提供している。
各サブクラスのインスタンスオブジェクトを作成するために cls() を利用することで、サブクラスごとの __init__() 呼び出しを無要にしている。
"""
def __init__(self, input_data: GenericInputData):
self.input_data = input_data
self.result = None

def map(self):
raise NotImplementedError

def reduce(self, other):
raise NotImplementedError

@classmethod
def create_workers(cls, input_class: GenericInputData, config: dict):
workers = []
for input_data in input_class.generate_inputs(config):
workers.append(cls(input_data))
return workers



GenericWorker サブクラスのインスタンス作成は、各クラスの cls() を利用してベースクラスのクラスメソッド内で完結しているため、派生クラスにおける変更の必要はありません。


class GenericLineCountWorker(GenericWorker):
"""
このプログラムにおける MapReduce モデル実装クラス
取得したデータ内に含まれる '\n' 文字の数をカウントする
"""
def map(self):
data = self.input_data.read()
self.result = data.count('\n')

def reduce(self, other: Worker):
self.result += other.result



全ての GenericWorker サブクラスには共通したインターフェース create_workers() を実装しましたから、利用するサブクラスが変更されても mapreduce() を書き換える必要は一切ありません。
完全にジェネリックになりました。


def generic_mapreduce(worker_class, input_class, config: dict) -> int:
workers = worker_class.create_workers(input_class, config)
return execute(workers)


config = {"data_dir": tempdir}
result = generic_mapreduce(GenericLineCountWorker, GenericPathInputData, config)

print(f"全てのファイルの行数は合計 {result} 行です。")
# 全てのファイルの行数は合計 4825 行です。



以前のバージョンとまったく同じ結果が取得できました。


mapreduce() を呼び出す際に必要な引数の数は増えましたが、それはよりジェネリックになった証でもあります。


今後他の GenericInputData サブクラス、GenericWorker サブクラスを作成し利用する場合でも、generic_mapreduce() に渡すパラメータを変更するだけであり、他のいかなる部分も変更する必要はありません。


まとめ:

1: Python でサポートされているコンストラクタは __init__() メソッドただ1つです。

2: @classmethod デコレータ ( decorator ) を利用することで、派生クラスごとに異なるコンストラクタ定義を記述できるようになります。

3: クラスメソッドポリモーフィズムを採用することで、サブクラス間の垣根をなくしよりジェネリックな方法での利用、インスタンス化をサポートできます。

この投稿をメールでシェアする

0 comments

コメントはまだありません。

コメントを追加する(不適切と思われるコメントは削除する場合があります)