Stone_Red/dvc
1
0
Fork 0
mirror of https://github.com/Stone-Red-Code/dvc.git synced 2026-11-02 07:18:25 +01:00
Code Issues Packages Projects Releases Wiki Activity

hdfs: fix drivers and tests

Browse Source 
Signed-off-by: Ruslan Kuprieiev <[email protected]>
This commit is contained in:
Ruslan Kuprieiev
2018-06-24 17:23:47 +03:00
parent b1558a9404
commit 71752f68a8
10 changed files with 58 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.travis.yml
-7
View File
@@ -53,13 +53,6 @@ before_install:
- which python
- which pip
- python --version
- if [[ "$TRAVIS_OS_NAME" == "osx" ]]; then
docker pull harisekhon/hadoop;
export HADOOP_CONTAINER_ID=$(docker run -dt harisekhon/hadoop);
echo $HADOOP_CONTAINER_ID;
export HADOOP_CONTAINER_IP=$(docker inspect -f '{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}' $HADOOP_CONTAINER_ID);
echo $HADOOP_CONTAINER_IP;
fi
install:
- bash ./scripts/ci/install.sh
script:
dvc/dependency/base.py
+6
View File
@@ -35,6 +35,12 @@ class DependencyBase(object):
def match(cls, url):
return re.match(cls.REGEX, url)
def group(self, name):
match = self.match(self.path)
if not match:
return None
return match.group(name)
@classmethod
def supported(cls, url):
return cls.match(url) != None
dvc/dependency/hdfs.py
+3 -1
View File
@@ -10,7 +10,7 @@ from dvc.remote.hdfs import RemoteHDFS
class DependencyHDFS(DependencyBase):
REGEX = r'^hdfs://.*$'
REGEX = RemoteHDFS.REGEX
def __init__(self, stage, path, info=None, remote=None):
super(DependencyHDFS, self).__init__(stage, path)
@@ -18,7 +18,9 @@ class DependencyHDFS(DependencyBase):
self.remote = remote if remote else RemoteHDFS(stage.project, {})
if remote:
path = posixpath.join(remote.url, urlparse(path).path.lstrip('/'))
user = remote.user if remote else self.group('user')
self.path_info = {'scheme': 'hdfs',
'user': user,
'url': path}
def changed(self):
dvc/dependency/ssh.py
-6
View File
@@ -29,12 +29,6 @@ class DependencySSH(DependencyBase):
'user': user,
'path': path}
def group(self, name):
match = self.match(self.path)
if not match:
return None
return match.group(name)
def changed(self):
return self.info != self.remote.save_info(self.path_info)
dvc/remote/base.py
+6
View File
@@ -46,6 +46,12 @@ class RemoteBase(object):
def match(cls, url):
return re.match(cls.REGEX, url)
def group(self, name):
m = self.match(self.url)
if not m:
return None
return m.group(name)
@staticmethod
def tmp_file(fname):
""" Temporary name for a partial download """
dvc/remote/hdfs.py
+30 -27
View File
@@ -1,5 +1,6 @@
import os
import re
import getpass
import posixpath
from subprocess import Popen, PIPE
@@ -15,23 +16,23 @@ from dvc.exceptions import DvcException
class RemoteHDFS(RemoteBase):
scheme='hdfs'
REGEX = r'^hdfs://(?P<path>.*)$'
REGEX = r'^hdfs://((?P<user>.*)@)?.*$'
PARAM_CHECKSUM = 'checksum'
def __init__(self, project, config):
self.project = project
self.url = config.get(Config.SECTION_REMOTE_URL, '/')
self.user = config.get(Config.SECTION_REMOTE_USER, None)
self.user = self.group('user')
if not self.user:
self.user = config.get(Config.SECTION_REMOTE_USER, getpass.getuser())
def hadoop_fs(self, cmd):
def hadoop_fs(self, cmd, user=None):
cmd = 'hadoop fs -' + cmd
env = os.environ.copy()
if self.user:
env['HADOOP_USER_NAME'] = self.user
if user:
cmd = 'HADOOP_USER_NAME={} '.format(user) + cmd
p = Popen(cmd,
shell=True,
close_fds=True,
env=env,
executable=os.getenv('SHELL'),
stdin=PIPE,
stdout=PIPE,
@@ -39,25 +40,25 @@ class RemoteHDFS(RemoteBase):
out, err = p.communicate()
if p.returncode != 0:
raise DvcException('HDFS command failed: {}: {}'.format(cmd, err))
return out
return out.decode('utf-8')
@staticmethod
def group(regex, s, gname):
def _group(regex, s, gname):
match = re.match(regex, s)
assert match != None
return match.group(gname)
def checksum(self, url):
def checksum(self, path_info):
regex = r'.*\t.*\t(?P<checksum>.*)'
stdout = self.hadoop_fs('checksum {}'.format(url))
return self.group(regex, stdout, 'checksum')
stdout = self.hadoop_fs('checksum {}'.format(path_info['url']), user=path_info['user'])
return self._group(regex, stdout, 'checksum')
def cp(self, from_url, to_url):
self.hadoop_fs('mkdir -p {}'.format(posixpath.dirname(to_url)))
self.hadoop_fs('cp {} {}'.format(from_url, to_url))
def cp(self, from_info, to_info):
self.hadoop_fs('mkdir -p {}'.format(posixpath.dirname(to_info['url'])), user=to_info['user'])
self.hadoop_fs('cp {} {}'.format(from_info['url'], to_info['url']), user=to_info['user'])
def rm(self, url):
self.hadoop_fs('rm {}'.format(url))
def rm(self, path_info):
self.hadoop_fs('rm {}'.format(path_info['url']), user=path_info['user'])
def save_info(self, path_info):
if path_info['scheme'] != 'hdfs':
@@ -65,7 +66,7 @@ class RemoteHDFS(RemoteBase):
assert path_info.get('url')
return {self.PARAM_CHECKSUM: self.checksum(path_info['url'])}
return {self.PARAM_CHECKSUM: self.checksum(path_info)}
def save(self, path_info):
if path_info['scheme'] != 'hdfs':
@@ -73,10 +74,11 @@ class RemoteHDFS(RemoteBase):
assert path_info.get('url')
checksum = self.checksum(path_info['url'])
dest_url = posixpath.join(self.url, checksum[0:2], checksum[2:])
checksum = self.checksum(path_info)
dest = path_info.copy()
dest['url'] = posixpath.join(self.url, checksum[0:2], checksum[2:])
self.cp(path_info['url'], dest_url)
self.cp(path_info, dest)
return {self.PARAM_CHECKSUM: checksum}
@@ -90,9 +92,10 @@ class RemoteHDFS(RemoteBase):
if not checksum:
return
url = posixpath.join(self.url, checksum[0:2], checksum[2:])
src = path_info.copy()
src['url'] = posixpath.join(self.url, checksum[0:2], checksum[2:])
self.cp(url, path_info['url'])
self.cp(src, path_info)
def remove(self, path_info):
if path_info['scheme'] != 'hdfs':
@@ -100,14 +103,14 @@ class RemoteHDFS(RemoteBase):
assert path_info.get('url')
self.rm(path_info['url'])
self.rm(path_info)
def upload(self, path, path_info):
if path_info['scheme'] != 'hdfs':
raise NotImplementedError
self.hadoop_fs('mkdir -p {}'.format(posixpath.dirname(path_info['url'])))
self.hadoop_fs('copyFromLocal {} {}'.format(path, path_info['url']))
self.hadoop_fs('mkdir -p {}'.format(posixpath.dirname(path_info['url'])), user=path_info['user'])
self.hadoop_fs('copyFromLocal {} {}'.format(path, path_info['url']), user=path_info['user'])
def download(self, path_info, path):
if path_info['scheme'] != 'hdfs':
@@ -117,4 +120,4 @@ class RemoteHDFS(RemoteBase):
if not os.path.exists(dname):
os.makedirs(dname)
self.hadoop_fs('copyToLocal {} {}'.format(path_info['url'], path))
self.hadoop_fs('copyToLocal {} {}'.format(path_info['url'], path), user=path_info['user'])
dvc/remote/ssh.py
-6
View File
@@ -52,12 +52,6 @@ class RemoteSSH(RemoteBase):
self.user = config.get(Config.SECTION_REMOTE_USER, getpass.getuser())
self.prefix = self.group('path')
def group(self, group):
m = self.match(self.url)
if not m:
return None
return m.group(group)
def cache_file_key(self, path):
relpath = os.path.relpath(os.path.abspath(path), self.project.cache.local.cache_dir)
return posixpath.join(self.prefix, relpath.replace('\\', '/'))
tests/basic_env.py
+2 -2
View File
@@ -16,9 +16,9 @@ class TestDir(TestCase):
DATA_CONTENTS = DATA
DATA_SUB_CONTENTS = DATA_SUB
FOO = 'foo'
FOO_CONTENTS = FOO
FOO_CONTENTS = FOO + '\n'
BAR = 'bar'
BAR_CONTENTS = BAR
BAR_CONTENTS = BAR + '\n'
CODE = 'code.py'
CODE_CONTENTS = 'import sys\nimport shutil\nshutil.copyfile(sys.argv[1], sys.argv[2])'
tests/log.log
+1
View File
@@ -0,0 +1 @@
/home/efiop/virtenvs/python3-dvc/bin/python: No module named tests.__main__; 'tests' is a package and cannot be directly executed
tests/test_repro.py
+10 -7
View File
@@ -293,7 +293,7 @@ class TestReproExternalBase(TestDvc):
out_foo_path = 'remote://' + remote_name + '/' + self.FOO
out_bar_path = bar_path
self.write(self.bucket, foo_key, 'foo')
self.write(self.bucket, foo_key, self.FOO_CONTENTS)
import_stage = self.dvc.imp(out_foo_path, 'import')
self.assertTrue(os.path.exists('import'))
@@ -303,7 +303,7 @@ class TestReproExternalBase(TestDvc):
deps=[out_foo_path],
cmd=self.cmd(foo_path, bar_path))
self.write(self.bucket, foo_key, 'bar')
self.write(self.bucket, foo_key, self.BAR_CONTENTS)
stages = self.dvc.reproduce(import_stage.path)
self.assertEqual(len(stages), 1)
@@ -372,14 +372,14 @@ class TestReproExternalHDFS(TestReproExternalBase):
@property
def bucket(self):
return os.getenv('HADOOP_CONTAINER_IP')
return 'root@' + os.getenv('HADOOP_CONTAINER_IP')
def cmd(self, i, o):
return 'HADOOP_USER_NAME=root hadoop fs -cp {} {}'.format(i, o)
def write(self, bucket, key, body):
url = self.scheme + '://' + bucket + '/' + key
p = Popen('hadoop fs -rm -f {}'.format(url),
p = Popen('HADOOP_USER_NAME=root hadoop fs -rm -f {}'.format(url),
shell=True,
executable=os.getenv('SHELL'),
stdin=PIPE,
@@ -387,13 +387,16 @@ class TestReproExternalHDFS(TestReproExternalBase):
stderr=PIPE)
p.communicate()
p = Popen('echo "{}" | hadoop fs -put - {}'.format(body, url),
p = Popen('echo "{}" | HADOOP_USER_NAME=root hadoop fs -put - {}'.format(body.strip(), url),
shell=True,
executable=os.getenv('SHELL'),
stdin=PIPE,
stdout=PIPE,
stderr=PIPE)
p.communicate()
out, err = p.communicate()
if p.returncode != 0:
print(out)
print(err)
self.assertEqual(p.returncode, 0)
@@ -438,7 +441,7 @@ class TestReproExternalSSH(TestReproExternalBase):
p.communicate()
self.assertEqual(p.returncode, 0)
p = Popen('echo "{}" | ssh {} "tr -d \'\\n\' > {}"'.format(body, dest, path),
p = Popen('echo "{}" | ssh {} "cat > {}"'.format(body.strip(), dest, path),
shell=True,
executable=os.getenv('SHELL'),
stdin=PIPE,