【Python】PyWebHdfsでHDFSを操作する

PythonでHDFSにファイルアップロードする処理を実装したのでメモ。

単純な処理ならcurlで十分なのですが、色々やりたい時は便利そうです。

事前にKerberos認証済みの想定です。

 ⬇️ まだ読めてないです😎

データウェブハウスツールキット―Web+データウェアハウスで最良のe‐ビジネス環境を構築

環境

やはりPythonのバージョンは3.7。

$ pip3 install pywebhdfs==0.4.1 requests-kerberos==0.12.0

最初にローカルでテストコードを実行しようとして、インストール出来なかったんですよね。

Mac osのバージョンによってはARCHFLAGSの宣言とか必要かも？

$ export ARCHFLAGS="-arch x86_64"
// 先にインストールが必要かも？
$ pip3 install pbr==5.5.1

PyWebHdfs

pythonhosted.org

実装

PyWebHdfsClientモジュールを継承して使いやすくしています。

# -*- coding: utf-8 -*-
from requests_kerberos import HTTPKerberosAuth, OPTIONAL
from pywebhdfs.webhdfs import PyWebHdfsClient


class HttpFsClient(PyWebHdfsClient):

    def __init__(self, base_uri_pattern, dir_permision='750', permision='640', overwrite=False,
                 reprication=3, block_size=134217728, stream=False, debug=False, verify=False, auth=None):
        self.dir_permision = dir_permision
        self.permision = permision
        self.overwrite = overwrite
        self.replication = replication
        self.block_size = block_size
        self.debug = debug
        if auth is None:
            auth = HTTPKerberosAuth(mutal_authentication=OPTIONAL)

        self.webhdfs_kwargs = {
            'base_uri_pattern': base_uri_pattern,
            'timeout': 120,
            'requests_extra_opts': {
                'verify': verify,
                'auth': auth,
                'stream': stream
            }
        }
        PyWebHdfsClient.__init__(self, **self.webhdfs_kwargs)

    def put_file(self, hdfs_path, src_file):
        with open(src_file, 'rb') as file_data:
            kwargs = {
                'overwrite': self.overwrite,
                'permision': self.permision,
                'blocksize': self.block_size,
                'reprication': self.replication
            }
            return PyWebHdfsClient.create_file(self, hdfs_path, file_data=file_data, **kwargs)

    def create_dir(self, hdfs_path):
        if not self.exists_file_dir(hdfs_path):
            self.make_dir(hdfs_path)


if __name__ == '__main__':
    client = HttpFsClient('https://xxxx.xxx:4443/webhdfs/v1/')
    client.create_dir('projects/sample/test')
    client.put_file('projects/sample/test/xxx.tsv')
    // 継承しているので、元のモジュールと同じように使える
    print(client.list_dir('projects/sample/test'))