<commit_before>from __future__ import absolute_import, division, print_function

import os
import re
import requests
import logging
from subprocess import Popen, PIPE

from .utils import conf_to_dict
from .env import CondaCreator
from .compatibility import FileNotFoundError, urlparse
from .exceptions import HDFSConfigException


logger = logging.getLogger(__name__)

JAR_FILE = "knit-1.0-SNAPSHOT.jar"
JAVA_APP = "io.continuum.knit.Client"


class Knit(object):
    """
    Connection to HDFS/YARN

    Parameters
    ----------
    nn: str
        Namenode hostname/ip
    nn: str
        Namenode hostname/ip
    nn_port: int
        Namenode Port (default: 9000)
    rm: str
        Resource Manager hostname
    rm_port: int
        Resource Manager port (default: 8088)
    autodetect: bool
        Autodetect NN/RM IP/Ports

    Examples
    --------

    >>> k = Knit()
    >>> app_id = k.start('sleep 100', num_containers=5, memory=1024)
    """
    def __init__(self, nn="localhost", nn_port=9000,
                 rm="localhost", rm_port=8088, autodetect=False):

        if autodetect:
             self.nn,  self.nn_port = self._yarn_conf(autodetect)
             self.rm,  self.rm_port = self._hdfs_conf(autodetect)

        self.nn = nn
        self.nn_port = str(nn_port)

        self.rm = rm
        self.rm_port = str(rm_port)

        self.java_lib_dir = os.path.join(os.path.dirname(__file__), "java_libs")
        self.KNIT_HOME = os.environ.get('KNIT_HOME') or self.java_lib_dir

        # must set KNIT_HOME ENV for YARN App
        os.environ['KNIT_HOME'] = self.KNIT_HOME
        self._hdfs_conf()
        self._yarn_conf()

    @property
    def JAR_FILE_PATH(self):
        return os.path.join(self.KNIT_HOME, JAR_FILE)

    def _yarn_conf(self, autodetect=False):
        """
        Load YARN config from default locations.
        Parameters
        ----------
        autodetect: bool

        Returns
        -------
            tuple (ip, port)

        """
        confd = os.environ.get('HADOOP_CONF_DIR', os.environ.get('HADOOP_INSTALL',
                               '') + '/hadoop/conf')
        afile = 'yarn-site.xml'
        conf = {}

        try:
            conf.update(conf_to_dict(os.sep.join([confd, afile])))
        except FileNotFoundError:
            pass
        if 'yarn.resourcemanager.webapp.address' in conf:
            url = conf['yarn.resourcemanager.webapp.address']
            u = urlparse(url)

            # handle host:port with no :// preabmle
            if u.path == url:
                conf['host'], conf['port'] = url.split(':')
            else:
                conf['host'] = u.hostname
                conf['port'] = u.port
        else:
            conf['host'] = "localhost"
            conf['port'] = "8088"

        if autodetect:
            return (conf['host'], conf['port'])

        if self.rm != conf['host']:
            msg = "Possible Resource Manager hostname mismatch.  Detected {}".format(conf['host'])
            raise HDFSConfigException(msg)
        if self.rm_port != conf['port']:
            msg = "Possible Resource Manager port mismatch.  Detected {}".format(conf['port'])
            raise HDFSConfigException(msg)

        return conf

    def _hdfs_conf(self, autodetect=False):
        """"
        Parameters
        ----------
        autodetect: bool

        Returns
        -------
            tuple (ip, port)

        """

        confd = os.environ.get('HADOOP_CONF_DIR', os.environ.get('HADOOP_INSTALL',
                               '') + '/hadoop/conf')
        files = 'core-site.xml', 'hdfs-site.xml'
        conf = {}
        for afile in files:
            try:
                conf.update(conf_to_dict(os.sep.join([confd, afile])))
            except FileNotFoundError:
                pass
        if 'fs.defaultFS' in conf:
            u = urlparse(conf['fs.defaultFS'])
            conf['host'] = u.hostname
            conf['port'] = u.port
        else:
            conf['host'] = "localhost"
            conf['port'] = "9000"

        if autodetect:
            return (conf['host'], conf['port'])

        if self.nn != conf['host']:
            msg = "Possible Namenode hostname mismatch.  Detected {}".format(conf['host'])
            raise HDFSConfigException(msg)
        if self.nn_port != str(conf['port']):
            msg = "Possible Namenode port mismatch.  Detected {}".format(conf['port'])
            raise HDFSConfigException(msg)

        return conf

    def start(self, cmd, num_containers=1, virtual_cores=1, memory=128, env=""):
        """
        Method to start a yarn app with a distributed shell

        Parameters
        ----------
        cmd: str
            command to run in each yarn container
        num_containers: int
            number of containers to start (default 1)
        num_containers: int
            Number of containers YARN should request (default: 1)
            * A container should be requested with the number of cores it can saturate, i.e.<commit_msg>small fix for autodetecting ip/ports<commit_after>from __future__ import absolute_import, division, print_function

import os
import re
import requests
import logging
from subprocess import Popen, PIPE

from .utils import conf_to_dict
from .env import CondaCreator
from .compatibility import FileNotFoundError, urlparse
from .exceptions import HDFSConfigException


logger = logging.getLogger(__name__)

JAR_FILE = "knit-1.0-SNAPSHOT.jar"
JAVA_APP = "io.continuum.knit.Client"


class Knit(object):
    """
    Connection to HDFS/YARN

    Parameters
    ----------
    nn: str
        Namenode hostname/ip
    nn: str
        Namenode hostname/ip
    nn_port: str
        Namenode Port (default: 9000)
    rm: str
        Resource Manager hostname
    rm_port: str
        Resource Manager port (default: 8088)
    autodetect: bool
        Autodetect NN/RM IP/Ports

    Examples
    --------

    >>> k = Knit()
    >>> app_id = k.start('sleep 100', num_containers=5, memory=1024)
    """
    def __init__(self, nn="localhost", nn_port=9000,
                 rm="localhost", rm_port=8088, autodetect=False):

        self.nn = nn
        self.nn_port = str(nn_port)

        self.rm = rm
        self.rm_port = str(rm_port)

        if autodetect:
            self.nn,  self.nn_port = self._hdfs_conf(autodetect)
            self.rm,  self.rm_port = self._yarn_conf(autodetect)
        else:
            # validates IP/Port is correct
            self._hdfs_conf()
            self._yarn_conf()

        self.java_lib_dir = os.path.join(os.path.dirname(__file__), "java_libs")
        self.KNIT_HOME = os.environ.get('KNIT_HOME') or self.java_lib_dir

        # must set KNIT_HOME ENV for YARN App
        os.environ['KNIT_HOME'] = self.KNIT_HOME

    @property
    def JAR_FILE_PATH(self):
        return os.path.join(self.KNIT_HOME, JAR_FILE)

    def _yarn_conf(self, autodetect=False):
        """
        Load YARN config from default locations.
        Parameters
        ----------
        autodetect: bool

        Returns
        -------
            tuple (ip, port)

        """
        confd = os.environ.get('HADOOP_CONF_DIR', os.environ.get('HADOOP_INSTALL',
                               '') + '/hadoop/conf')
        afile = 'yarn-site.xml'
        conf = {}

        try:
            conf.update(conf_to_dict(os.sep.join([confd, afile])))
        except FileNotFoundError:
            pass
        if 'yarn.resourcemanager.webapp.address' in conf:
            url = conf['yarn.resourcemanager.webapp.address']
            u = urlparse(url)

            # handle host:port with no :// preabmle
            if u.path == url:
                conf['host'], conf['port'] = url.split(':')
            else:
                conf['host'] = u.hostname
                conf['port'] = u.port
        else:
            conf['host'] = "localhost"
            conf['port'] = "8088"

        if autodetect:
            return (conf['host'], conf['port'])

        if self.rm != conf['host']:
            msg = "Possible Resource Manager hostname mismatch.  Detected {}".format(conf['host'])
            raise HDFSConfigException(msg)
        if str(self.rm_port) != str(conf['port']):
            msg = "Possible Resource Manager port mismatch.  Detected {}".format(conf['port'])
            raise HDFSConfigException(msg)

        return conf

    def _hdfs_conf(self, autodetect=False):
        """"
        Parameters
        ----------
        autodetect: bool

        Returns
        -------
            tuple (ip, port)

        """

        confd = os.environ.get('HADOOP_CONF_DIR', os.environ.get('HADOOP_INSTALL',
                               '') + '/hadoop/conf')
        afile = 'core-site.xml'
        conf = {}
        try:
            conf.update(conf_to_dict(os.sep.join([confd, afile])))
        except FileNotFoundError:
            pass
        
        if 'fs.defaultFS' in conf:
            u = urlparse(conf['fs.defaultFS'])
            conf['host'] = u.hostname
            conf['port'] = u.port
        else:
            conf['host'] = "localhost"
            conf['port'] = "9000"

        if autodetect:
            return (conf['host'], conf['port'])

        if self.nn != conf['host']:
            msg = "Possible Namenode hostname mismatch.  Detected {}".format(conf['host'])
            raise HDFSConfigException(msg)

        if str(self.nn_port) != str(conf['port']):
            msg = "Possible Namenode port mismatch.  Detected {}".format(conf['port'])
            raise HDFSConfigException(msg)

        return conf

    def start(self, cmd, num_containers=1, virtual_cores=1, memory=128, env=""):
        """
        Method to start a yarn app with a distributed shell

        Parameters
        ----------
        cmd: str
            command to run in each yarn container
        num_containers: int
            number of containers to start (default 1)
        num_containers: int
            Number of containers YARN should request (default: 1)
            * A container should be requested with the number of cores it can saturate, i.e.
{'fraction_non_alphanumeric': 0.12033195020746888, 'fraction_numerical': 0.008101165777514326, 'mean_word_length': 1.851549295774648, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 5}, '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': '6905632', 'n_tokens_mistral': 3162, 'n_tokens_neox': 2989, 'n_words': 721}