Airflow 를 사용하여 데이터 파이프 라인을 조정하고 있습니다. 작업 중 하나 RouteModel에서 S3에서 절인 객체 ( 인스턴스) 를로드하려고합니다 .
def read_file_from_s3(bucket, file):
from inference.route_model import RouteModel
s3_loader = S3Client(bucket, None)
buffer = s3_loader.get_file(file)
data = pickle.loads(buffer.read())
이 오류가 발생합니다.
Traceback (most recent call last):
File "/Users/cyrusghazanfar/Desktop/startup-studio/pilota_project/pilota_ml/env/lib/python3.6/site-packages/airflow/models/taskinstance.py", line 926, in _run_raw_task
result = task_copy.execute(context=context)
File "/Users/cyrusghazanfar/Desktop/startup-studio/pilota_project/pilota_ml/env/lib/python3.6/site-packages/airflow/operators/python_operator.py", line 113, in execute
return_value = self.execute_callable()
File "/Users/cyrusghazanfar/Desktop/startup-studio/pilota_project/pilota_ml/env/lib/python3.6/site-packages/airflow/operators/python_operator.py", line 118, in execute_callable
return self.python_callable(*self.op_args, **self.op_kwargs)
File "/Users/cyrusghazanfar/Desktop/startup-studio/pilota_project/pilota_ml/inference/predict.py", line 43, in get_pred_for_flight
pred_state, pred_state_prob, pred_dt = tst_pipeline.get_prediction(format_pred_od)
File "/Users/cyrusghazanfar/Desktop/startup-studio/pilota_project/pilota_ml/inference/pipeline.py", line 174, in get_prediction
route_model = self.rm_loader.get_model(self.rm_dict[r_key]['rm_key'])
File "/Users/cyrusghazanfar/Desktop/startup-studio/pilota_project/pilota_ml/inference/dataloader.py", line 40, in get_model
route_model = read_file_from_s3(self.loc, fname)
File "/Users/cyrusghazanfar/Desktop/startup-studio/pilota_project/pilota_ml/inference/dataloader.py", line 96, in read_file_from_s3
data = pickle.loads(buffer.read())
AttributeError: Can't get attribute 'RouteModel' on <module '__main__' from '/Users/cyrusghazanfar/Desktop/startup-studio/pilota_project/pilota_ml/env/bin/airflow'>
커스텀 클래스로 작업 할 때 피클되는 클래스는 피클을 읽는 프로세스의 네임 스페이스에 나타나야합니다.이 경우에는 Airflow입니다.
노트:
파일 절임 방식을 변경할 수 없습니다.
도와주세요 :)
이 문제를 해결하려면 피클 파일이 참조하는 특정 인스턴스의 사용자 정의 클래스를 명시 적으로 반환하는 사용자 정의 언 피클 러를 작성해야했습니다.
class CustomUnpickler(pickle.Unpickler):
def find_class(self, module, name):
if name == 'RouteModel':
from inference.route_model import RouteModel
return RouteModel
return super().find_class(module, name)
data = CustomUnpickler(io.BytesIO(buffer.read())).load()
이 기사는 인터넷에서 수집됩니다. 재 인쇄 할 때 출처를 알려주십시오.
침해가 발생한 경우 연락 주시기 바랍니다[email protected] 삭제
몇 마디 만하겠습니다