2

Hi I'm running a task using the PythonOperator. It appears that the task was actually able to run just fine and the returned value was what I expected (it is a large XML output from an API call). However, I am getting a ERROR - (builtins.RecursionError) maximum recursion depth exceeded in comparison. My python callable is returning a value so I'm assuming there is an XCOM push and that it is trying to serialize the output to be ingested by subsequent operators. But I'm not sure exactly how to resolve since I don't see a configuration to either a) increase the recursion depth of the Pickle serializer (suggested here) or 2) any error handling during an XCOM push

My full trace is below

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
Minh
  • 2,180
  • 5
  • 23
  • 50

1 Answers1

2

Found out that there is a limit for pickling Python objects set by the database's BLOB or BINARY LARGE OBJECT. To get around this you can

  • try using Fileflow
  • dump the file into a temporary folder and push the filepath via the XCOM functionality
  • Handle the entire process within a single task and push certain values via XCOM
Minh
  • 2,180
  • 5
  • 23
  • 50