title: Convert python script to airflow dag
body: <p>I have identified the below script as being really useful for anyone running Amazon Redshift:</p>
  <pre><code>#!/usr/bin/env python
  from __future__ import print_function
  
  '''
  analyze-vacuum-schema.py
  * Copyright 2015, Amazon.com, Inc. or its affiliates. All Rights Reserved.
  *
  * Licensed under the Amazon Software License (the &quot;License&quot;).
  * You may not use this file except in compliance with the License.
  * A copy of the License is located at
  *
  * http://aws.amazon.com/asl/
  *
  * or in the &quot;license&quot; file accompanying this file. This file is distributed
  * on an &quot;AS IS&quot; BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
  * express or implied. See the License for the specific language governing
  * permissions and limitations under the License.
  The Redshift Analyze Vacuum Utility gives you the ability to automate VACUUM and ANALYZE operations.
  When run, it will analyze or vacuum an entire schema or individual tables. This Utility Analyzes
  and Vacuums table(s) in a Redshift Database schema, based on certain parameters like unsorted,
  stats off and size of the table and system alerts from stl_explain &amp; stl_alert_event_log.
  By turning on/off '--analyze-flag' and  '--vacuum-flag' parameters, you can run it as  'vacuum-only'
  or  'analyze-only' utility. This script can be scheduled to run VACUUM and ANALYZE as part of
  regular maintenance/housekeeping activities, when there are less database activities (quiet period).
  This script will:
     1) Analyze a single table or tables in a schema based on,
          a) Alerts from stl_explain &amp; stl_alert_event_log.
          b) 'stats off' metrics from SVV_TABLE_INFO.
     2) Vacuum a single table or tables in a schema based on,
          a) The alerts from stl_alert_event_log.
          b) The 'unsorted' and 'size' metrics from SVV_TABLE_INFO.
          c) Vacuum reindex to analyze the interleaved sort keys
  Srinikri Amazon Web Services (2015)
  11/21/2015 : Added support for vacuum reindex to analyze the interleaved sort keys.
  09/01/2017 : Fixed issues with interleaved sort key tables per https://github.com/awslabs/amazon-redshift-utils/issues/184
  11/09/2017 : Refactored to support running in AWS Lambda
  14/12/2017 : Refactored to support a more sensible interface style with kwargs
  '''
  import os
  import sys
  import argparse
  
  # add the lib directory to the sys path
  try:
      sys.path.append(os.path.join(os.path.dirname(__file__), &quot;lib&quot;))
      sys.path.append(os.path.join(os.path.dirname(__file__), &quot;..&quot;))
  except:
      pass
  
  import getopt
  import analyze_vacuum
  import config_constants
  
  __version__ = &quot;.9.2.1&quot;
  
  OK = 0
  ERROR = 1
  INVALID_ARGS = 2
  NO_WORK = 3
  TERMINATED_BY_USER = 4
  NO_CONNECTION = 5
  
  # setup cli args
  parser = argparse.ArgumentParser()
  parser.add_argument(&quot;--analyze-flag&quot;, dest=&quot;analyze_flag&quot;, default=True, type=bool,
                      help=&quot;Flag to turn ON/OFF ANALYZE functionality (True or False : Default = True &quot;)
  parser.add_argument(&quot;--max-unsorted-pct&quot;, dest=&quot;max_unsorted_pct&quot;,
                      help=&quot;Maximum unsorted percentage(% to consider a table for vacuum : Default = 50%&quot;)
  parser.add_argument(&quot;--min-interleaved-cnt&quot;, dest=&quot;min_interleaved_cnt&quot;, type=int,
                      help=&quot;Minimum stv_interleaved_counts records to consider a table for vacuum reindex: Default = 0&quot;)
  parser.add_argument(&quot;--min-interleaved-skew&quot;, dest=&quot;min_interleaved_skew&quot;,
                      help=&quot;Minimum index skew to consider a table for vacuum reindex: Default = 1.4&quot;)
  parser.add_argument(&quot;--min-unsorted-pct&quot;, dest=&quot;min_unsorted_pct&quot;,
                      help=&quot;Minimum unsorted percentage(% to consider a table for vacuum : Default = 5%&quot;)
  parser.add_argument(&quot;--stats-off-pct &quot;, dest=&quot;stats_off_pct&quot;,
                      help=&quot;Minimum stats off percentage(% to consider a table for analyze : Default = 10%&quot;)
  parser.add_argument(&quot;--table-name&quot;, dest=&quot;table_name&quot;,
                      help=&quot;A specific table to be Analyzed or Vacuumed if analyze-schema is not desired&quot;)
  parser.add_argument(&quot;--vacuum-flag&quot;, dest=&quot;vacuum_flag&quot;, default=True, type=bool,
                      help=&quot;Flag to turn ON/OFF VACUUM functionality (True or False :  Default = True&quot;)
  parser.add_argument(&quot;--vacuum-parameter&quot;, dest=&quot;vacuum_parameter&quot;,
                      help=&quot;Vacuum parameters [ FULL | SORT ONLY | DELETE ONLY | REINDEX ] Default = FULL&quot;)
  parser.add_argument(&quot;--blacklisted-tables&quot;, dest=&quot;blacklisted_tables&quot;, help=&quot;The tables we do not want to Vacuum&quot;)
  parser.add_argument(&quot;--db-conn-opts&quot;, dest=&quot;db_conn_opts&quot;,
                      help=&quot;Additional connection options. name1=opt1[ name2=opt2]..&quot;)
  parser.add_argument(&quot;--db-host&quot;, dest=&quot;db_host&quot;, required=True, help=&quot;The Cluster endpoint&quot;)
  parser.add_argument(&quot;--db-port&quot;, dest=&quot;db_port&quot;, type=int, required=True,
                      help=&quot;The Cluster endpoint port : Default = 5439&quot;)
  parser.add_argument(&quot;--db-pwd&quot;, dest=&quot;db_pwd&quot;, help=&quot;The Password for the Database User to connect to&quot;)
  parser.add_argument(&quot;--db-user&quot;, dest=&quot;db_user&quot;, required=True, help=&quot;The Database User to connect to&quot;)
  parser.add_argument(&quot;--debug &quot;, dest=&quot;debug&quot;, default=False,
                      help=&quot;Generate Debug Output including SQL Statements being run&quot;)
  parser.add_argument(&quot;--ignore-errors&quot;, dest=&quot;ignore_errors&quot;, default=True,
                      help=&quot;Ignore errors raised when running and continue processing&quot;)
  parser.add_argument(&quot;--max-table-size-mb&quot;, dest=&quot;max_table_size_mb&quot;, type=int,
                      help=&quot;Maximum table size in MB : Default = 700*1024 MB&quot;)
  parser.add_argument(&quot;--output-file&quot;, dest=&quot;output_file&quot;, help=&quot;The full path to the output file to be generated&quot;)
  parser.add_argument(&quot;--predicate-cols&quot;, dest=&quot;predicate_cols&quot;, help=&quot;Analyze predicate columns only&quot;)
  parser.add_argument(&quot;--query-group&quot;, dest=&quot;query_group&quot;, help=&quot;Set the query_group for all queries&quot;)
  parser.add_argument(&quot;--require-ssl&quot;, dest=&quot;require_ssl&quot;, default=False,
                      help=&quot;Does the connection require SSL? (True | False&quot;)
  parser.add_argument(&quot;--schema-name&quot;, dest=&quot;schema_name&quot;,
                      help=&quot;The Schema to be Analyzed or Vacuumed (REGEX: Default = public&quot;)
  parser.add_argument(&quot;--slot-count&quot;, dest=&quot;slot_count&quot;, help=&quot;Modify the wlm_query_slot_count : Default = 1&quot;)
  parser.add_argument(&quot;--suppress-cloudwatch&quot;, dest=&quot;suppress_cw&quot;,
                      help=&quot;Don't emit CloudWatch metrics for analyze or vacuum when set to True&quot;)
  parser.add_argument(&quot;--db&quot;, dest=&quot;db&quot;, help=&quot;The Database to Use&quot;)
  full_args = parser.parse_args()
  parse_args = {}
  # remove args that end up as None
  for k, v in vars(full_args).items():
      if v is not None:
          parse_args[k] = v
  
  
  def main(argv):
      # get environmental args
      args = {config_constants.DB_NAME: os.environ.get('PGDATABASE', None),
              config_constants.DB_USER: os.environ.get('PGUSER', None),
              config_constants.DB_PASSWORD: os.environ.get('PGPASSWORD', None),
              config_constants.DB_HOST: os.environ.get('PGHOST', None),
              config_constants.DB_PORT: os.environ.get('PGPORT', 5439)}
  
      # add argparse args
      args.update(parse_args)
  
      if args.get(config_constants.OUTPUT_FILE) is not None:
          sys.stdout = open(args.get(config_constants.OUTPUT_FILE), 'w')
  
      # invoke the main method of the utility
      result = analyze_vacuum.run_analyze_vacuum(**args)
  
      if result is not None:
          sys.exit(result)
      else:
          sys.exit(0)
  
  
  if __name__ == &quot;__main__&quot;:
      main(sys.argv)
  </code></pre>
  <p>However, you can only &quot;easily&quot; schedule it from cron in an EC2 or similar scheduler. So, I have been trying to find a way to run it as an airflow dag.</p>
  <p>I have found two similar questions on stackoverflow and I guess it is missing some import commands perhaps? like the below:</p>
  <pre><code>from airflow import DAG
  from airflow.models import Variable
  </code></pre>
  <p>I'm hoping to find someone familiar enough with airflow to be able to help me find the required entries or point me in the right direction?</p>
  <p>If so, I can create a fork from the original script:</p>
  <p><a href="https://github.com/awslabs/amazon-redshift-utils/blob/master/src/AnalyzeVacuumUtility/analyze-vacuum-schema.py" rel="nofollow noreferrer">https://github.com/awslabs/amazon-redshift-utils/blob/master/src/AnalyzeVacuumUtility/analyze-vacuum-schema.py</a></p>
  <p>Which will assist others in the future with the same goal.</p>