Adding mysql_postoperator to HiveToMySqlTransfer

airflow/operators/hive_to_mysql.py

@@ -25,9 +25,14 @@ class HiveToMySqlTransfer(BaseOperator):
         coming in, allowing the task to be idempotent (running the task
         twice won't double load data)
     :type mysql_preoperator: str
+    :param mysql_postoperator: sql statement to run against mysql after the
+        import, typically used to move data from staging to production
+        and issue cleanup commands.
+    :type mysql_postoperator: str
     """
 
-    template_fields = ('sql', 'mysql_table', 'mysql_preoperator')
+    template_fields = ('sql', 'mysql_table', 'mysql_preoperator',
+        'mysql_postoperator')
     template_ext = ('.sql',)
     ui_color = '#a0e08c'
 
@@ -39,12 +44,14 @@ def __init__(
             hiveserver2_conn_id='hiveserver2_default',
             mysql_conn_id='mysql_default',
             mysql_preoperator=None,
+            mysql_postoperator=None,
             *args, **kwargs):
         super(HiveToMySqlTransfer, self).__init__(*args, **kwargs)
         self.sql = sql
         self.mysql_table = mysql_table
         self.mysql_conn_id = mysql_conn_id
         self.mysql_preoperator = mysql_preoperator
+        self.mysql_postoperator = mysql_postoperator
         self.hiveserver2_conn_id = hiveserver2_conn_id
 
     def execute(self, context):
@@ -61,3 +68,8 @@ def execute(self, context):
 
         logging.info("Inserting rows into MySQL")
         mysql.insert_rows(table=self.mysql_table, rows=results)
+
+        if self.mysql_postoperator:
+            logging.info("Running MySQL postoperator")
+            logging.info(self.mysql_postoperator)
+            mysql.run(self.mysql_postoperator)