#NEWLINE# def run_cli( self , pig , verbose = True ) : #NEWLINE# #INDENT# '\n        Run an pig script using the pig cli\n\n        >>> ph = PigCliHook()\n        >>> result = ph.run_cli("ls /;")\n        >>> ("hdfs://" in result)\n        True\n        ' #NEWLINE# with TemporaryDirectory ( prefix = 'airflow_pigop_' ) as tmp_dir : #NEWLINE# #INDENT# with NamedTemporaryFile ( dir = tmp_dir ) as f : #NEWLINE# #INDENT# f . write ( pig ) #NEWLINE# f . flush ( ) #NEWLINE# fname = f . name #NEWLINE# pig_bin = 'pig' #NEWLINE# cmd_extra = [ ] #NEWLINE# pig_cmd = ( [ pig_bin , '-f' , fname ] + cmd_extra ) #NEWLINE# if self . pig_properties : #NEWLINE# #INDENT# pig_properties_list = self . pig_properties . split ( ) #NEWLINE# pig_cmd . extend ( pig_properties_list ) #NEWLINE# #UNINDENT# if verbose : #NEWLINE# #INDENT# logging . info ( ' ' . join ( pig_cmd ) ) #NEWLINE# #UNINDENT# sp = subprocess . Popen ( self , stdout = subprocess . PIPE , stderr = subprocess . STDOUT , cwd = tmp_dir ) #NEWLINE# self . sp = sp #NEWLINE# stdout = '' #NEWLINE# for line in iter ( sp . stdout . readline , '' ) : #NEWLINE# #INDENT# stdout += line #NEWLINE# if verbose : #NEWLINE# #INDENT# logging . info ( line . strip ( ) ) #NEWLINE# #UNINDENT# #UNINDENT# sp . wait ( ) #NEWLINE# if sp . returncode : #NEWLINE# #INDENT# raise AirflowException ( stdout ) #NEWLINE# #UNINDENT# return stdout