python - "Cannot pickle files that are not opened for reading"使用 Client.map()

Question

我正在尝试使用 dask.distributed 根据多个 CSV 文件的内容同时更新 Postgresql 数据库。理想情况下，我们将 CSV 文件分发给 N 个工作人员，每个工作人员会将 CSV 文件内容插入数据库。但是，在将任务分配给工作人员时使用 Client.map() 时，我们会遇到 Cannot pickle files that notopened for read 异常。

这是代码的精简版本:

def _work(csv_path):
   db = Database() # encapsulates interaction w/ postgresql database
   db.open()

   count = 0

   with csv_path.open('r') as csv_file:
       reader = csv.DictReader(csv_file)

       for record in reader:
           db.insert(record)
           count += 1

   db.close()

   return count


client = Client(processes=False)

csv_files = Path('/data/files/').glob('*.csv')

csv_futures = client.map(_work, csv_files) # error occurs here

for finished in  as_completed(csv_futures):
   count = finished.result()
   print(count)

根据相关的stackoverflow和github问题，我成功使用cloudpickle对函数和参数进行序列化和反序列化。

cloudpickle.loads(cloudpickle.dumps(_work))
Out[69]: <function _work(csv_path)>

和

files = list(Path('/data/files/').glob('*.csv'))
files
Out[73]: 
[PosixPath('/data/files/208.csv'),
 PosixPath('/data/files/332.csv'),
 PosixPath('/data/files/125.csv'),
 PosixPath('/data/files/8.csv')]
cloudpickle.loads(cloudpickle.dumps(files))
Out[74]: 
[PosixPath('/data/files/208.csv'),
 PosixPath('/data/files/332.csv'),
 PosixPath('/data/files/125.csv'),
 PosixPath('/data/files/8.csv')]

所以，问题出在其他地方。

Answer 1

确切的异常(exception)是这样的:

  File "/Users/may/anaconda/envs/eagle-i/lib/python3.6/site-packages/cloudpickle/cloudpickle.py", line 841, in save_file
    raise pickle.PicklingError("Cannot pickle files that are not opened for reading: %s" % obj.mode)
_pickle.PicklingError: Cannot pickle files that are not opened for reading: a

单步执行调试器，我很好奇 obj 是什么？是，这是这样的:

<_io.TextIOWrapper name='/tmp/logs/ei_sched.log' mode='a' encoding='UTF-8'>

在上面给出的示例代码片段中，我省略了对记录器的调用，cloudpickle 就是这样的。正在提示。在尝试使用 dask 并行化此功能之前，日志记录是此功能的剩余工件。一旦我从传递给 Client.map() 的函数中删除了日志记录调用， ，事情按预期进行。

顺便说一句，这是 cloudpickle 的一个很好的收获。因为日志记录到单个文件不应该由 dask 工作人员完成。