<commit_before># Copyright (c) 2012 Spotify AB
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not
# use this file except in compliance with the License. You may obtain a copy of
# the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations under
# the License.

import subprocess
import os
import random
import tempfile
import urlparse
import luigi.format
import datetime

def use_cdh4_syntax():
    """
    CDH4 (hadoop 2+) has a slightly different syntax for interacting with
    hdfs via the command line. The default version is CDH4, but one can
    override this setting with "cdh3" in the hadoop section of the config in
    order to use the old syntax
    """
    import interface
    return interface.get_config().get("hadoop", "version", "cdh4").lower() == "cdh4"

def exists(path):
    cmd = ['hadoop', 'fs', '-test', '-e', path]
    p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
    stdout, _ = p.communicate()

    if stdout:
        # TODO: Having certain stuff on the classpath might trigger this case.
        # Ideally we should be able to ignore it, because apparently there is
        # also some case where data on stdout signals an error.
        # From @freider:
        # If I remember correctly there are some cases where
        # `hadoop fs -test` exits with exit status 0 (which would mean
        # the file exists) although the file doesn't exist, if there is some
        # special error. We ran into this, that's why we added detection of
        # output of the command.
        # Afaicr this is a known bug but it hadn't been fixed in our version
        # of hadoop.
        raise RuntimeError("Command %r failed [exit code %d] because it wrote output.\n---Output---\n%s\n------------" % (cmd, p.returncode, stdout))
    elif p.returncode not in (0, 1):
        raise RuntimeError("Command %r failed with return code %s" % (cmd, p.returncode))

    if p.returncode == 0:
        return True
    elif p.returncode == 1:
        return False
    assert False


def tmppath(path=None):
    return tempfile.gettempdir() + '/' + (path + "-" if path else "") + "luigitemp-%08d" % random.randrange(1e9)


def rename(path, dest):
    parent_dir = os.path.dirname(dest)
    if parent_dir != '' and not exists(parent_dir):
        mkdir(parent_dir)
    cmd = ['hadoop', 'fs', '-mv', path, dest]
    if subprocess.call(cmd):
        raise RuntimeError('Command %s failed' % repr(cmd))


def remove(path, recursive=True):
    if recursive:
        if use_cdh4_syntax():
            cmd = ['hadoop', 'fs', '-rm', '-r', path]
        else:
            cmd = ['hadoop', 'fs', '-rmr', path]
    else:
        cmd = ['hadoop', 'fs', '-rm', path]
    if subprocess.call(cmd):
        raise RuntimeError('Command %s failed' % repr(cmd))


def mkdir(path):
    cmd = ['hadoop', 'fs', '-mkdir', path]
    if subprocess.call(cmd):
        raise RuntimeError('Command %s failed' % repr(cmd))


def listdir(path, ignore_directories=False, ignore_files=False,
            include_size=False, include_type=False, include_time=False):
    if not path:
        path = "."  # default to current/home catalog

    cmd = ['hadoop', 'fs', '-ls', path]
    proc = subprocess.Popen(cmd, stdout=subprocess.PIPE)
    lines = proc.stdout

    for line in lines:
        if line.startswith('Found'):
            continue  # "hadoop fs -ls" outputs "Found %d items" as its first line
        line = line.rstrip("\n")
        if ignore_directories and line[0] == 'd':
            continue
        if ignore_files and line[0] == '-':
            continue
        data = line.split(' ')

        file = data[-1]
        size = int(data[-4])
        line_type = line[0]
        extra_data = ()

        if include_size:
            extra_data += (size,)
        if include_type:
            extra_data += (line_type,)
        if include_time:
            time_str = '%sT%s' % (data[-3], data[-2])
            modification_time = datetime.datetime.strptime(time_str,
                                                           '%Y-%m-%dT%H:%M')
            extra_data += (modification_time,)

        if len(extra_data) > 0:
            yield (file,) + extra_data
        else:
            yield file


class HdfsReadPipe(luigi.format.InputPipeProcessWrapper):
    def __init__(self, path):
        super(HdfsReadPipe, self).__init__(['hadoop', 'fs', '-cat', path])


class HdfsAtomicWritePipe(luigi.format.OutputPipeProcessWrapper):
    """ File like object for writing to HDFS

    The referenced file is first written to a temporary location and then
    renamed to final location on close(). If close() isn't called
    the temporary file will be cleaned up when this object is
    garbage collected

    TODO: if this is buggy, change it so it first writes to a
    local temporary file and then uploads it on completion
    """

    def __init__(self, path):
        self.path = path
        self.tmppath = tmppath(self.path)
        tmpdir = os.path.dirname(self.tmppath)
        if use_cdh4_syntax():<commit_msg>Moved hdfs operations into HdfsClient class, added aliases so they can be used as module methods as before.<commit_after># Copyright (c) 2012 Spotify AB
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not
# use this file except in compliance with the License. You may obtain a copy of
# the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations under
# the License.

import subprocess
import os
import random
import tempfile
import urlparse
import luigi.format
import datetime

def use_cdh4_syntax():
    """
    CDH4 (hadoop 2+) has a slightly different syntax for interacting with
    hdfs via the command line. The default version is CDH4, but one can
    override this setting with "cdh3" in the hadoop section of the config in
    order to use the old syntax
    """
    import interface
    return interface.get_config().get("hadoop", "version", "cdh4").lower() == "cdh4"


def tmppath(path=None):
    return tempfile.gettempdir() + '/' + (path + "-" if path else "") + "luigitemp-%08d" % random.randrange(1e9)


class HdfsClient(object):
    def exists(self, path):
        cmd = ['hadoop', 'fs', '-test', '-e', path]
        p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
        stdout, _ = p.communicate()
        if stdout and 'No such file or directory' not in stdout:
            # TODO2: Using globs with -test -e will print 'No such file or directory'
            # when no files exists
            # TODO: Having certain stuff on the classpath might trigger this case.
            # Ideally we should be able to ignore it, because apparently there is
            # also some case where data on stdout signals an error.
            # From @freider:
            # If I remember correctly there are some cases where
            # `hadoop fs -test` exits with exit status 0 (which would mean
            # the file exists) although the file doesn't exist, if there is some
            # special error. We ran into this, that's why we added detection of
            # output of the command.
            # Afaicr this is a known bug but it hadn't been fixed in our version
            # of hadoop.
            raise RuntimeError("Command %r failed [exit code %d] because it wrote output.\n---Output---\n%s\n------------" % (cmd, p.returncode, stdout))
        elif p.returncode not in (0, 1):
            raise RuntimeError("Command %r failed with return code %s" % (cmd, p.returncode))

        if p.returncode == 0:
            return True
        elif p.returncode == 1:
            return False
        assert False


    def rename(self, path, dest):
        parent_dir = os.path.dirname(dest)
        if parent_dir != '' and not exists(parent_dir):
            mkdir(parent_dir)
        cmd = ['hadoop', 'fs', '-mv', path, dest]
        if subprocess.call(cmd):
            raise RuntimeError('Command %s failed' % repr(cmd))


    def remove(self, path, recursive=True):
        if recursive:
            if use_cdh4_syntax():
                cmd = ['hadoop', 'fs', '-rm', '-r', path]
            else:
                cmd = ['hadoop', 'fs', '-rmr', path]
        else:
            cmd = ['hadoop', 'fs', '-rm', path]
        if subprocess.call(cmd):
            raise RuntimeError('Command %s failed' % repr(cmd))


    def mkdir(self, path):
        cmd = ['hadoop', 'fs', '-mkdir', path]
        if subprocess.call(cmd):
            raise RuntimeError('Command %s failed' % repr(cmd))


    def listdir(self, path, ignore_directories=False, ignore_files=False,
                include_size=False, include_type=False, include_time=False):
        if not path:
            path = "."  # default to current/home catalog

        cmd = ['hadoop', 'fs', '-ls', path]
        proc = subprocess.Popen(cmd, stdout=subprocess.PIPE)
        lines = proc.stdout

        for line in lines:
            if line.startswith('Found'):
                continue  # "hadoop fs -ls" outputs "Found %d items" as its first line
            line = line.rstrip("\n")
            if ignore_directories and line[0] == 'd':
                continue
            if ignore_files and line[0] == '-':
                continue
            data = line.split(' ')

            file = data[-1]
            size = int(data[-4])
            line_type = line[0]
            extra_data = ()

            if include_size:
                extra_data += (size,)
            if include_type:
                extra_data += (line_type,)
            if include_time:
                time_str = '%sT%s' % (data[-3], data[-2])
                modification_time = datetime.datetime.strptime(time_str,
                                                               '%Y-%m-%dT%H:%M')
                extra_data += (modification_time,)

            if len(extra_data) > 0:
                yield (file,) + extra_data
            else:
                yield file

client = HdfsClient()

exists = client.exists
rename = client.rename
remove = client.remove
mkdir = client.mkdir
listdir = client.listdir

class HdfsReadPipe(luigi.format.InputPipeProcessWrapper):
    def __init__(self, path):
        super(HdfsReadPipe, self).__init__(['hadoop', 'fs', '-cat', path])


class HdfsAtomicWritePipe(luigi.format.OutputPipeProcessWrapper):
    """ File like object for writing to HDFS

    The referenced file is first written to a temporary location and then
    renamed to final location on close(). If close() isn't called
    the temporary file will be cleaned up when this object is
    garbage collected

    TODO: if this is buggy, change it so it first writes to a
    local temporary file and then uploads it on completion
    """

    def __init__(self, path):
        self.path = path
        self.tmppath = tmppath(self.path)
        tmpdir = os.path.dirname(self.tmppath)
        if use_cdh4_syntax():
{'fraction_non_alphanumeric': 0.10679189494738672, 'fraction_numerical': 0.006000521784503, 'mean_word_length': 2.3684827182190977, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3983551', 'n_tokens_mistral': 3346, 'n_tokens_neox': 3221, 'n_words': 1195}