#NEWLINE# def execute( self , context ) : #NEWLINE# #INDENT# hive = HiveServer2Hook ( hiveserver2_conn_id = self . hiveserver2_conn_id ) #NEWLINE# logging . info ( 'Extracting data from Hive' ) #NEWLINE# logging . info ( self . sql ) #NEWLINE# if self . bulk_load : #NEWLINE# #INDENT# tmpfile = NamedTemporaryFile ( ) #NEWLINE# hive . to_csv ( tmpfile . sql , tmpfile . name , delimiter = '\t' , lineterminator = '\n' , output_header = False ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# results = hive . get_records ( self . sql ) #NEWLINE# #UNINDENT# mysql = MySqlHook ( mysql_conn_id = self . mysql_conn_id ) #NEWLINE# if self . mysql_preoperator : #NEWLINE# #INDENT# logging . info ( 'Running MySQL preoperator' ) #NEWLINE# mysql . run ( self . mysql_preoperator ) #NEWLINE# #UNINDENT# logging . info ( 'Inserting rows into MySQL' ) #NEWLINE# if self . bulk_load : #NEWLINE# #INDENT# mysql . bulk_load ( table = self . mysql_table , tmp_file = tmpfile . name ) #NEWLINE# tmpfile . close ( ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# mysql . insert_rows ( table = self . mysql_table , rows = results ) #NEWLINE# #UNINDENT# if self . mysql_postoperator : #NEWLINE# #INDENT# logging . info ( 'Running MySQL postoperator' ) #NEWLINE# mysql . run ( self . mysql_postoperator ) #NEWLINE# #UNINDENT# logging . info ( 'Done.' )