走读中学到的技巧

Posted 2020-11-24 testzcy

自动发掘dag实例
    def process_file(self, filepath):
        mod_name, file_ext = os.path.splitext(os.path.split(filepath)[-1])
        if file_ext != ‘.py‘:
            return
        try:
            logging.info("Importing " + filepath)
            m = imp.load_source(mod_name, filepath)
        except :
            traceback.print_exc()
            logging.error("Failed to import: " + filepath)
        else:
            for dag in m.__dict__.values():
                if type(dag) == DAG:
                    if dag.dag_id in self.dags:
                        raise Exception(
                            ‘Two DAGs with the same dag_id. No good.‘)
                    self.dags[dag.dag_id] = dag
                    dag.dagbag = self