はじめに

近年、オープンソースソフトウェア（Open-source software, 略称: OSS）が幅広く使われるようになってきました。自分が所属しているデータサイエンスチームでもビックデータ処理や機械学習にOSSを活用しています。その中には、幾つかGitHubで上位にランキングされているOSSもあります。

 
このように多くのユーザーから利用されてまた多くのコントリビューターによって開発されているOSSは、多くの場合に開発・補修が安定的に行われるので安心して使う事が可能です。しかし、実際には数少ないコントリビューターによって開発されているケースが圧倒的に多く、使用している最中にバグに出会うこともあります。
今回は、hdfs3 (Python HDFS client library) を使った時に遭遇した実際の経験を元に、バグ発見・修正を通じてOSSにコントリビュートする方法を紹介したいと思います。

物事の始まり

データサイエンスチームでは社内外から得られた大規模データを一つの場所に集めて管理・利用しています。Datalakeとも呼ばれるこの仕組みは異質のデータを統合的に活用するビックデータ処理に当たってとても有効であり、そのためにHDFS (Hadoop分散ファイルシステム)を利用しています。そしてデータ処理には分散処理エンジンであるApache Sparkを利用します。
しかし、サンプルデータを利用した探索的データ解析を行う時には逆にこの仕組みが不便になることが時々あります。たとえば、普段は使わないプログラムやライブラリが必要な場合、クラスタ上のワーカーノード (実際分散処理が行われるサーバ群)にこれらを設置する必要があるからです。また、分散処理プログラミングでは一般的なプログラミングよりデバギングが難しいです。
このような理由で、自分は探索的データ解析を行う時にはPython (とJupyter notebook)を利用しています。そしてHDFS上にあるデータをいちいちローカルに持ってくることは非効率的なので、直接データにアクセスするためにhdfs3ライブラリを使っていました。
たとえば、下記のような方法でHDFS上のファイルを直接読み込むことが可能です。

from hdfs3 import HDFileSystem
host = "namenode1.datascience.data-hotel.net"
port = 8020
hdfs = HDFileSystem(host=host, port=port)
with hdfs.open('/user/data/mySample.txt') as f:
    for byteline in f:
        # do something here

 
データサイエンスチームで使用しているHadoop環境は障害対応としてHigh Availability (HA)を導入しています。そのため、接続時にhostとport情報が下記のように変わります。hdfs3ライブラリはHadoop HDFS環境ファイル (hdfs-site.xml)を参考し、接続を行います。

from hdfs3 import HDFileSystem
# host: HDFS Nameservice.
# port: Don't use port for HA-mode HDFS.
host = "researchcluster.datascience"
hdfs = HDFileSystem(host=host)
with hdfs.open('/user/data/mySample.txt') as f:
    for byteline in f:
        # do something here

なんと！下記のようなエラーで接続ができませんでした。

/home/myaccount/.conda/envs/py3/lib/python3.6/site-packages/hdfs3/core.py:127: UserWarning: Setting conf parameter port failed
warnings.warn('Setting conf parameter %s failed' % par)

OSSコントリビューションの流れ

自分のコード・実行環境に問題がなければ、次はOSSライブラリの問題 (バグ)である可能性があります。ここでは下記の三つの段階でバグを解決し、またOSSへコントリビューションをする方法を説明します。

1. 原因の追跡

最初にエラーの原因が本当にOSSライブラリのバグなのかを確かめる必要があります。エラーメッセージで問題になってたhdfs3/core.py:127のソースコードを見てみましょう。

    def connect(self):
        """ Connect to the name node
        This happens automatically at startup
        """
        get_lib()
        conf = self.conf.copy()
        if self._handle:
            return
        if HDFileSystem._first_pid is None:
            HDFileSystem._first_pid = os.getpid()
        elif HDFileSystem._first_pid != os.getpid():
            warnings.warn("Attempting to re-use hdfs3 in child process %d, "
                          "but it was initialized in parent process %d. "
                          "Beware that hdfs3 is not fork-safe and this may "
                          "lead to bugs or crashes."
                          % (os.getpid(), HDFileSystem._first_pid),
                          RuntimeWarning, stacklevel=2)
        o = _lib.hdfsNewBuilder()
        if conf['port'] is not None:
            _lib.hdfsBuilderSetNameNodePort(o, conf.pop('port'))
        _lib.hdfsBuilderSetNameNode(o, ensure_bytes(conf.pop('host')))
        if 'user' in conf:
            _lib.hdfsBuilderSetUserName(
                o, ensure_bytes(conf.pop('user')))
        if 'ticket_cache' in conf:
            _lib.hdfsBuilderSetKerbTicketCachePath(
                o, ensure_bytes(conf.pop('ticket_cache')))
        if 'token' in conf:
            _lib.hdfsBuilderSetToken(o, ensure_bytes(conf.pop('token')))
        for par, val in conf.items():
            if not _lib.hdfsBuilderConfSetStr(o, ensure_bytes(par),
                                              ensure_bytes(val)) == 0:
                warnings.warn('Setting conf parameter %s failed' % par)
        fs = _lib.hdfsBuilderConnect(o)
        _lib.hdfsFreeBuilder(o)
        if fs:
            logger.debug("Connect to handle %d", fs.contents.filesystem)
            self._handle = fs
        else:
            msg = ensure_string(_lib.hdfsGetLastError()).split('\n')[0]
            raise ConnectionError('Connection Failed: {}'.format(msg))

127行では与えられたすべてのパラメーターの設定を行っています。一見何の問題もないように見えます。その上の部分も見てみると、host, port, user, ticket_cache, tokenの五つのパラメータはそれぞれ違うメソッドを使って設定を行うことがわかります。整理すると下記のようになります。

• port – _lib.hdfsBuilderSetNameNodePort()
• host – _lib.hdfsBuilderSetNameNode()
• user – _lib.hdfsBuilderSetUserName()
• ticket_cache – _lib.hdfsBuilderSetKerbTicketCachePath()
• token – _lib.hdfsBuilderSetToken()
• その他 – _lib.hdfsBuilderConfSetStr()

気づきましたか？　この部分は何かおかしいですね。
五つのパラメータは既にそれぞれ違うメソッドを使って設定されてますが、124-127行のfor loopではそれらがまだ含まれているconfオブジェクトを使って設定を行っています。
ここで問題になる可能性がある部分はportだけです。
hostは無条件で設定されるので2回実行されても同じ結果になります。user, ticket_cache, tokenはconfオブジェクトに存在する時だけ設定を行うためこちらも2回実行されても問題ないです。しかし、portの場合、None値を持っている時は設定メソッドを呼んではいけない(110行)ですが、124-127では無条件で実行しています。
このバグを直すため下記のように変更して既に設定されたパラメータを2回設定しないようにしました。

        for par, val in conf.items():
            if par in ['port', 'user', 'ticket_cache', 'token']:
                continue
            if not _lib.hdfsBuilderConfSetStr(o, ensure_bytes(par),
                                              ensure_bytes(val)) == 0:
                warnings.warn('Setting conf parameter %s failed' % par)

2. バグの報告・バグフィックスの提供

いくつかのテストを終えてOSSライブラリのバグであることが確実になると、開発者へ報告を行います。hdfs3はGitHub上で管理されているのでIssue掲示板でバグの報告をします。
実際のやりとりがdask/hdfs3 issue 132で確認できます。
既存のコントリビューターからのコメントを受けて修正内容を精錬していきます。この時、元のレポジトリを自分のアカウントにforkしてそこに修正内容を反映していくと便利です。
最終的に変更されたコードは下記のようになります。if-elseではなく、上のブロックで使われたパラメータをpopしました。最初のコードより読みやすくなってます。

        _lib.hdfsBuilderSetNameNode(o, ensure_bytes(conf.pop('host')))
        port = conf.pop('port', None)
        if port is not None:
            _lib.hdfsBuilderSetNameNodePort(o, port)
        user = conf.pop('user', None)
        if user is not None:
            _lib.hdfsBuilderSetUserName(o, ensure_bytes(user))
        ticket_cache = conf.pop('ticket_cache', None)
        if ticket_cache is not None:
            _lib.hdfsBuilderSetKerbTicketCachePath(o, ensure_bytes(ticket_cache))
        token = conf.pop('token', None)
        if token is not None:
            _lib.hdfsBuilderSetToken(o, ensure_bytes(token))
        for par, val in conf.items():
            if not _lib.hdfsBuilderConfSetStr(o, ensure_bytes(par),
                                              ensure_bytes(val)) == 0:
                warnings.warn('Setting conf parameter %s failed' % par)

3. OSSコントリビューターへの一歩、Pull Request

コントリビューターから同意が得られたら修正内容をupstream (元のレポジトリ)へ反映させる段階に進めます。自分のアカウント上にforkしたレポジトリに修正内容を反映し、Pull Requestを送ります。Upstreamのコントリビューターから承認が降りれば修正内容が反映されます。
これでOSSコントリビューターとして小さいですが大きな一歩を踏み出せることができました。
 

おわりに

OSSへのコントリビューションは様々な形で行われます。今回のブログではユーザーとしてOSSのバグに遭遇した時、バグフィックスを提供することでコントリビューションする方法を説明しました。
自分が使うOSSにコントリビューションするという意味ではどんなに小さなものでも大きな意味を持つのではないでしょうか。