0
안녕하세요 저는 PythonOperator를 사용하여 작업을 실행하고 있습니다. 이 작업은 실제로 실제로 잘 돌아갈 수 있었고 반환 된 값은 예상 한 것입니다 (API 호출의 큰 XML 출력 임). 그러나 ERROR - (builtins.RecursionError) maximum recursion depth exceeded in comparison이 표시됩니다. 내 파이썬 호출 가능 값을 반환하므로 XCOM 푸시가 있고 후속 연산자에 의해 처리되도록 출력을 직렬화하려고합니다. 그러나 Pickle serializer (here 제안)의 재귀 깊이를 높이거나 XCOM 푸시 중에 오류 처리를 구성하는 방법이 보이지 않기 때문에 해결 방법을 정확히 알지 못합니다.공기 흐름 : XCOM Inseert 중 Pickle Depth Recursion
내 전체 추적
INFO - Subtask: [2017-11-08 14:00:14,545] {models.py:1342} INFO - Executing <Task(PythonOperator): test_task_xml> on 2017-11-07 00:00:00
INFO - Subtask: [2017-11-08 14:00:31,817] {python_operator.py:81} INFO - Done. Returned value was: <QueryResult><Query><Answer>12345</Answer> ... (12321456 characters truncated) ... </Query></QueryResult>
INFO - Subtask: [2017-11-08 14:00:31,839] {models.py:1417} ERROR - (builtins.RecursionError) maximum recursion depth exceeded in comparison [SQL: 'INSERT INTO xcom (key, value, timestamp, execution_date, task_id, dag_id) VALUES (%(key)s, %(value)s, now(), %(execution_date)s, %(task_id)s, %(dag_id)s) RETURNING xcom.id'] [parameters: [{'dag_id': 'test_dag', 'key': 'return_value', 'value': <QueryResult><Query><Answer>12345</Answer> ... (12321456 characters truncated) ... </Query></QueryResult>, 'task_id': 'test_task_xml', 'execution_date': datetime.datetime(2017, 11, 7, 0, 0)}]]
INFO - Subtask: Traceback (most recent call last):
INFO - Subtask: File "/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1116, in _execute_context
INFO - Subtask: context = constructor(dialect, self, conn, *args)
INFO - Subtask: File "/lib/python3.6/site-packages/sqlalchemy/engine/default.py", line 690, in _init_compiled
INFO - Subtask: for key in compiled_params
INFO - Subtask: File "/lib/python3.6/site-packages/sqlalchemy/engine/default.py", line 690, in <genexpr>
INFO - Subtask: for key in compiled_params
INFO - Subtask: File "/lib/python3.6/site-packages/sqlalchemy/sql/sqltypes.py", line 1516, in process
INFO - Subtask: value = dumps(value, protocol)
INFO - Subtask: dump(obj, file, protocol, byref, fmode, recurse)#, strictio)
INFO - Subtask: File "/lib/python3.6/site-packages/dill/dill.py", line 274, in dump
[INFO - Subtask: pik.dump(obj)
INFO - Subtask: File "/lib/python3.6/pickle.py", line 409, in dump
INFO - Subtask: self.save(obj)
INFO - Subtask: File "/lib/python3.6/pickle.py", line 521, in save
INFO - Subtask: self.save_reduce(obj=obj, *rv)
INFO - Subtask: File "/lib/python3.6/pickle.py", line 634, in save_reduce
INFO - Subtask: save(state)
INFO - Subtask: File "/lib/python3.6/pickle.py", line 476, in save
INFO - Subtask: f(self, obj) # Call unbound method with explicit self
INFO - Subtask: File "/lib/python3.6/site-packages/dill/dill.py", line 871, in save_module_dict
INFO - Subtask: StockPickler.save_dict(pickler, obj)
INFO - Subtask: File "/lib/python3.6/pickle.py", line 821, in save_dict
INFO - Subtask: self._batch_setitems(obj.items())
INFO - Subtask: File "/lib/python3.6/pickle.py", line 847, in _batch_setitems
INFO - Subtask: save(v)
INFO - Subtask: File "/lib/python3.6/pickle.py", line 476, in save
INFO - Subtask: f(self, obj) # Call unbound method with explicit self
INFO - Subtask: File "/lib/python3.6/pickle.py", line 781, in save_list
INFO - Subtask: self._batch_appends(obj)
INFO - Subtask: File "/lib/python3.6/pickle.py", line 805, in _batch_appends
INFO - Subtask: save(x)
INFO - Subtask: File "/lib/python3.6/pickle.py", line 521, in save
INFO - Subtask: self.save_reduce(obj=obj, *rv)
INFO - Subtask: File "/lib/python3.6/pickle.py", line 634, in save_reduce
INFO - Subtask: save(state)
INFO - Subtask: File "/lib/python3.6/pickle.py", line 476, in save
INFO - Subtask: f(self, obj) # Call unbound method with explicit self
INFO - Subtask: File "/lib/python3.6/site-packages/dill/dill.py", line 871, in save_module_dict