<s> """<STR_LIT>""" <EOL> from __future__ import division <EOL> import os <EOL> import shutil <EOL> from vistrails . core . modules . basic_modules import File , Boolean , String , PathObject , Path <EOL> from vistrails . core . modules . config import IPort , OPort , ModuleSettings <EOL> from vistrails . core . modules . vistrails_module import ModuleError <EOL> from base import HadoopBaseModule <EOL> class HDFSPut ( HadoopBaseModule ) : <EOL> """<STR_LIT>""" <EOL> _settings = ModuleSettings ( namespace = '<STR_LIT>' ) <EOL> _input_ports = [ IPort ( '<STR_LIT>' , File ) , <EOL> IPort ( '<STR_LIT>' , String ) , <EOL> IPort ( '<STR_LIT>' , Boolean ) , <EOL> IPort ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> _output_ports = [ OPort ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> OPort ( '<STR_LIT>' , String ) ] <EOL> def __init__ ( self ) : <EOL> HadoopBaseModule . __init__ ( self ) <EOL> def compute ( self ) : <EOL> machine = self . get_machine ( ) <EOL> jm = self . job_monitor ( ) <EOL> id = self . signature <EOL> job = jm . getCache ( id ) <EOL> if not job : <EOL> remote = self . get_input ( '<STR_LIT>' ) <EOL> local = self . get_input ( '<STR_LIT>' ) <EOL> override = self . force_get_input ( '<STR_LIT>' , False ) <EOL> if '<STR_LIT>' not in remote : <EOL> remote = self . add_prefix ( remote , machine ) <EOL> if not int ( self . call_hdfs ( '<STR_LIT>' + remote + <EOL> '<STR_LIT>' , machine ) ) : <EOL> if override : <EOL> self . call_hdfs ( '<STR_LIT>' + remote , machine ) <EOL> else : <EOL> raise ModuleError ( self , '<STR_LIT>' ) <EOL> tempfile = machine . remote . send_command ( '<STR_LIT>' ) . strip ( ) <EOL> result = machine . sendfile ( local . name , tempfile ) <EOL> self . call_hdfs ( '<STR_LIT>' % ( tempfile , remote ) , machine ) <EOL> result = machine . remote . rm ( tempfile , force = True , recursively = True ) <EOL> d = { '<STR_LIT>' : remote , '<STR_LIT>' : local . name } <EOL> self . set_job_machine ( d , machine ) <EOL> jm . setCache ( id , d , self . job_name ( ) ) <EOL> job = jm . getJob ( id ) <EOL> self . set_output ( '<STR_LIT>' , job . parameters [ '<STR_LIT>' ] ) <EOL> self . set_output ( '<STR_LIT>' , machine ) <EOL> class HDFSGet ( HadoopBaseModule ) : <EOL> """<STR_LIT>""" <EOL> _settings = ModuleSettings ( namespace = '<STR_LIT>' ) <EOL> _input_ports = [ IPort ( '<STR_LIT>' , Path ) , <EOL> IPort ( '<STR_LIT>' , String ) , <EOL> IPort ( '<STR_LIT>' , Boolean ) , <EOL> IPort ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> _output_ports = [ OPort ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> OPort ( '<STR_LIT>' , File ) ] <EOL> def __init__ ( self ) : <EOL> HadoopBaseModule . __init__ ( self ) <EOL> def compute ( self ) : <EOL> machine = self . get_machine ( ) <EOL> jm = self . job_monitor ( ) <EOL> id = self . signature <EOL> job = jm . getCache ( id ) <EOL> if not job : <EOL> remote = self . get_input ( '<STR_LIT>' ) <EOL> local = self . get_input ( '<STR_LIT>' ) <EOL> override = self . force_get_input ( '<STR_LIT>' , False ) <EOL> if '<STR_LIT>' not in remote : <EOL> remote = self . add_prefix ( remote , machine ) <EOL> if os . path . exists ( local . name ) : <EOL> if override == False : <EOL> raise ModuleError ( self , '<STR_LIT>' ) <EOL> else : <EOL> if os . path . isdir ( local . name ) : <EOL> shutil . rmtree ( local . name ) <EOL> else : <EOL> os . unlink ( local . name ) <EOL> tempfile = machine . remote . send_command ( '<STR_LIT>' ) . strip ( ) <EOL> result = self . call_hdfs ( '<STR_LIT>' % ( remote , tempfile ) , machine ) <EOL> result = machine . local . send_command ( '<STR_LIT>' % local . name ) <EOL> result = machine . sync ( local . name , <EOL> tempfile , <EOL> mode = machine . MODE_REMOTE_LOCAL , <EOL> use_tar = True ) <EOL> result = machine . remote . rm ( tempfile , force = True , recursively = True ) <EOL> d = { '<STR_LIT>' : remote , '<STR_LIT>' : local . name } <EOL> self . set_job_machine ( d , machine ) <EOL> jm . setCache ( id , d , self . job_name ( ) ) <EOL> job = jm . getCache ( id ) <EOL> self . set_output ( '<STR_LIT>' , PathObject ( job . parameters [ '<STR_LIT>' ] ) ) <EOL> self . set_output ( '<STR_LIT>' , machine ) <EOL> class HDFSEnsureNew ( HadoopBaseModule ) : <EOL> """<STR_LIT>""" <EOL> _settings = ModuleSettings ( namespace = '<STR_LIT>' ) <EOL> _input_ports = [ IPort ( '<STR_LIT:Name>' , String ) , <EOL> IPort ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> _output_ports = [ OPort ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> OPort ( '<STR_LIT:Name>' , String ) ] <EOL> def __init__ ( self ) : <EOL> HadoopBaseModule . __init__ ( self ) <EOL> def compute ( self ) : <EOL> machine = self . get_machine ( ) <EOL> jm = self . job_monitor ( ) <EOL> id = self . signature <EOL> job = jm . getCache ( id ) <EOL> if not job : <EOL> entry_name = self . get_input ( '<STR_LIT:Name>' ) <EOL> if '<STR_LIT>' not in entry_name : <EOL> entry_name = self . add_prefix ( entry_name , machine ) <EOL> if not int ( self . call_hdfs ( '<STR_LIT>' + entry_name + <EOL> '<STR_LIT>' , machine ) ) : <EOL> self . call_hdfs ( '<STR_LIT>' + entry_name , machine ) <EOL> d = { '<STR_LIT>' : entry_name } <EOL> self . set_job_machine ( d , machine ) <EOL> jm . setCache ( id , d , self . job_name ( ) ) <EOL> job = jm . getCache ( id ) <EOL> self . set_output ( '<STR_LIT:Name>' , job . parameters [ '<STR_LIT>' ] ) <EOL> self . set_output ( '<STR_LIT>' , machine ) <EOL> def register ( ) : <EOL> return [ HDFSPut , HDFSGet , HDFSEnsureNew ] </s>
