Python operator 생성 후 import 사용하려고하는데 설정이 잘 안되네요.

Question

안녕하세요.Python operator 생성 후 import시 &#34;cannot import name &#34; Error이 발생합니다. 우선, 설정은 아래와 같습니다.1. airflow.cfg 파일의 plugins 폴더 경로# This path must be absolute dags_folder &#61; /home/postgres/airflow/ldas/dags # Where your Airflow plugins are storedplugins_folder &#61; /home/postgres/airflow/ldas/plugins 2. /home/postgres/airflow/ldas/plugins 의 ldas_plugin.py 내용# 기존에 CheckFileOperator, InitOperator, QueryOperator 는 존재하였고, 현재 TangoCheckFileOperator를 추가하려고 from __future__ import division, absolute_import, print_function import loggingfrom os import path from airflow import AirflowExceptionfrom airflow.hooks.postgres_hook import PostgresHookfrom airflow.models import BaseOperatorfrom airflow.operators.postgres_operator import PostgresOperatorfrom airflow.operators.python_operator import PythonOperatorfrom airflow.plugins_manager import AirflowPlugin#from airflow.utils.decorators import apply_defaultsfrom airflow.utils import apply_defaults log &#61; logging.getLogger(__name__) class CheckFileOperator(BaseOperator): template_fields&#61;('filename',) &#64;apply_defaults def __init__(self, filename, *args, **kwargs): super(CheckFileOperator, self).__init__(*args, **kwargs) self.filename &#61; filename # self.execute_date &#61; kwargs.get('templates_dict').get('execute_date') # self.execute_date &#61; execute_date def execute(self, context): # log.info('called CheckFileOperator2 ' &#43; self.execute_date) log.info('filename : ' &#43; self.filename) if path.isfile(self.filename ) is True and path.getsize(self.filename ) > 0: # if path.isfile(self.filename ) is True: return 'File exists' else: if path.isfile(self.filename ) is True and path.getsize(self.filename ) &#61;&#61; 0: raise AirflowException('File size is zero') else: raise AirflowException('File no exists') class TangoCheckFileOperator(BaseOperator): template_fields&#61;('filename',) &#64;apply_defaults def __init__(self, filename, *args, **kwargs): super(TangoCheckFileOperator, self).__init__(*args, **kwargs) self.filename &#61; filename def execute(self, context): log.info('filename : ' &#43; self.filename) if path.isfile(self.filename ) is True and path.getsize(self.filename ) > 0: return 'File exists' else: if path.isfile(self.filename ) is True and path.getsize(self.filename ) &#61;&#61; 0: raise AirflowException('File size is zero') else: raise AirflowException('File no exists') class InitOperator(BaseOperator): &#64;apply_defaults def __init__(self, data, *args, **kwargs): super(InitOperator, self).__init__(*args, **kwargs) self.data &#61; data def execute(self, context): if self.data: print(self.data) else: raise AirflowException('no data') class QueryOperator(PostgresOperator): &#64;apply_defaults def __init__(self, sql, *args, **kwargs): self.sql &#61; sql self.postgres_conn_id &#61; 'QGIS_Postgres' def execute(self, context): self.log.info('Executing: %s', self.sql) self.hook &#61; PostgresHook(postgres_conn_id&#61;self.postgres_conn_id, schema&#61;'QGIS' ) self.hook.run(self.sql, self.autocommit, parameters&#61;self.parameters) for output in self.hook.conn.notices: self.log.info(output) class LDASPlugin(AirflowPlugin): name &#61; &#34;ldas_plugin&#34; #operators &#61; [CheckFileOperator, InitOperator] operators &#61; [CheckFileOperator, TangoCheckFileOperator, InitOperator, QueryOperator] # A list of class(es) derived from BaseHook hooks &#61; [] # A list of class(es) derived from BaseExecutor executors &#61; [] # A list of references to inject into the macros namespace macros &#61; [] # A list of objects created from a class derived # from flask_admin.BaseView admin_views &#61; [] # A list of Blueprint object created from flask.Blueprint flask_blueprints &#61; [] # A list of menu links (flask_admin.base.MenuLink) menu_links &#61; [] 3. /home/postgres/airflow/ldas/dags 의 daily_insert.py 내용 AS_IS : from airflow.operators.ldas_plugin import CheckFileOperator, InitOperatorTO_BE : from airflow.operators.ldas_plugin import CheckFileOperator, TangoCheckFileOperator, InitOperator TangoCheckFileOperator를 사용하려고 import 선언시 Broken DAG: [/home/postgres/airflow/ldas/dags/daily_insert.py] cannot import name 'TangoCheckFileOperator' from 'airflow.operators.ldas_plugin' (unknown location) 에러가 뜨는데 어떻게 해결을 해야할까요.

Python operator 생성 후 import 사용하려고하는데 설정이 잘 안되네요.

0개의 답변

Q&A 목록