<s> """<STR_LIT>""" <EOL> __all__ = [ '<STR_LIT>' ] <EOL> from docopt import docopt <EOL> from os import getcwd , pardir , sep <EOL> from os . path import abspath , basename , exists , isabs , relpath <EOL> from time import sleep <EOL> from . . job import Job <EOL> from . . project import Project <EOL> from . . remote import Execution , Session <EOL> from . . util import ( AzkabanError , Config , catch , suppress_urllib_warnings , <EOL> temppath ) <EOL> import logging as lg <EOL> import sys <EOL> class PigJob ( Job ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , * options ) : <EOL> super ( PigJob , self ) . __init__ ( <EOL> { '<STR_LIT:type>' : Config ( ) . get_option ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) } , <EOL> * options <EOL> ) <EOL> try : <EOL> self . path = self . options [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> raise AzkabanError ( '<STR_LIT>' ) <EOL> else : <EOL> self . options [ '<STR_LIT>' ] = self . path . lstrip ( '<STR_LIT:/>' ) <EOL> self . join_prefix ( '<STR_LIT>' , '<STR_LIT:U+0020>' , '<STR_LIT>' ) <EOL> def on_add ( self , project , name , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if not '<STR_LIT>' in kwargs : <EOL> project . add_file ( self . path ) <EOL> class _PigProject ( Project ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , name , paths , pig_type = None ) : <EOL> super ( _PigProject , self ) . __init__ ( name , register = False , root = getcwd ( ) ) <EOL> self . ordered_jobs = [ basename ( path ) for path in paths ] <EOL> for path , dep in zip ( paths , [ None ] + self . ordered_jobs ) : <EOL> options = { '<STR_LIT>' : abspath ( path ) } <EOL> if pig_type : <EOL> options [ '<STR_LIT:type>' ] = pig_type <EOL> options [ '<STR_LIT>' ] = dep or '<STR_LIT>' <EOL> self . add_job ( basename ( path ) , PigJob ( options ) ) <EOL> def logs ( self , execution , delay = <NUM_LIT:10> ) : <EOL> """<STR_LIT>""" <EOL> ok_statuses = set ( [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> for job in self . ordered_jobs : <EOL> while execution . status [ '<STR_LIT:status>' ] == '<STR_LIT>' : <EOL> self . _logger . warning ( '<STR_LIT>' , job ) <EOL> sleep ( delay ) <EOL> for line in execution . job_logs ( job ) : <EOL> yield line <EOL> if not execution . status [ '<STR_LIT:status>' ] in ok_statuses : <EOL> raise AzkabanError ( '<STR_LIT>' , job ) <EOL> else : <EOL> self . _logger . info ( '<STR_LIT>' , job ) <EOL> @ catch ( AzkabanError ) <EOL> def main ( ) : <EOL> """<STR_LIT>""" <EOL> args = docopt ( __doc__ ) <EOL> cfg = Config ( ) <EOL> logger = lg . getLogger ( ) <EOL> logger . setLevel ( lg . DEBUG ) <EOL> handler = cfg . get_file_handler ( '<STR_LIT>' ) <EOL> if handler : <EOL> logger . addHandler ( handler ) <EOL> suppress_urllib_warnings ( ) <EOL> if args [ '<STR_LIT>' ] : <EOL> if handler : <EOL> sys . stdout . write ( '<STR_LIT>' % ( handler . baseFilename , ) ) <EOL> sys . exit ( <NUM_LIT:0> ) <EOL> else : <EOL> raise AzkabanError ( '<STR_LIT>' ) <EOL> paths = args [ '<STR_LIT>' ] <EOL> jars = args [ '<STR_LIT>' ] or [ ] <EOL> if args [ '<STR_LIT>' ] : <EOL> session = Session ( url = args [ '<STR_LIT>' ] , config = cfg ) <EOL> else : <EOL> alias = args [ '<STR_LIT>' ] or cfg . get_option ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> session = Session . from_alias ( alias = alias , config = cfg ) <EOL> project = _PigProject ( <EOL> name = args [ '<STR_LIT>' ] or cfg . get_option ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> paths = paths , <EOL> pig_type = args [ '<STR_LIT>' ] , <EOL> ) <EOL> project . properties = { <EOL> '<STR_LIT>' : session . user , <EOL> '<STR_LIT>' : '<STR_LIT:U+002C>' . join ( abspath ( j ) . lstrip ( sep ) for j in jars ) , <EOL> } <EOL> project . properties . update ( <EOL> opt . split ( '<STR_LIT:=>' , <NUM_LIT:1> ) <EOL> for opt in args [ '<STR_LIT>' ] <EOL> if '<STR_LIT:=>' in opt <EOL> ) <EOL> job_paths = ( opt for opt in args [ '<STR_LIT>' ] if not '<STR_LIT:=>' in opt ) <EOL> for i , job_path in enumerate ( job_paths ) : <EOL> if not exists ( job_path ) : <EOL> raise AzkabanError ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> % ( job_path , ) <EOL> ) <EOL> project . add_file ( abspath ( job_path ) , '<STR_LIT>' % ( i , ) ) <EOL> for fpath in args [ '<STR_LIT>' ] : <EOL> if exists ( fpath ) and not isabs ( fpath ) and pardir in relpath ( fpath ) : <EOL> raise AzkabanError ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> % ( fpath , ) <EOL> ) <EOL> project . add_file ( fpath ) <EOL> for jar in jars : <EOL> project . add_file ( abspath ( jar ) ) <EOL> with temppath ( ) as tpath : <EOL> project . build ( tpath ) <EOL> while True : <EOL> try : <EOL> session . upload_project ( project , tpath ) <EOL> except AzkabanError : <EOL> session . create_project ( project , project ) <EOL> else : <EOL> break <EOL> res = session . run_workflow ( project , basename ( paths [ - <NUM_LIT:1> ] ) ) <EOL> exe = Execution ( session , res [ '<STR_LIT>' ] ) <EOL> if args [ '<STR_LIT>' ] : <EOL> sys . stdout . write ( '<STR_LIT>' % ( exe . url , ) ) <EOL> else : <EOL> try : <EOL> for line in project . logs ( exe ) : <EOL> sys . stdout . write ( '<STR_LIT>' % ( line . encode ( '<STR_LIT:utf-8>' ) , ) ) <EOL> except KeyboardInterrupt : <EOL> choice = raw_input ( '<STR_LIT>' ) . lower ( ) <EOL> if choice and choice [ <NUM_LIT:0> ] == '<STR_LIT:y>' : <EOL> sys . stdout . write ( '<STR_LIT>' ) <EOL> exe . cancel ( ) <EOL> sys . stdout . write ( '<STR_LIT>' ) <EOL> else : <EOL> sys . stdout . write ( '<STR_LIT>' % ( exe . url , ) ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> main ( ) </s>
