dask + luigi: 提高 ValueError('url type not understood: %s' % urlpath)

Question

我正在尝试将 dask 与 luigi 合并，虽然业务逻辑本身运行良好，但当我运行一个 Luigi 任务时，代码开始抛出错误：

raise ValueError('url type not understood: %s' % urlpath)
ValueError: url type not understood: <_io.TextIOWrapper name='../data/2017_04_11_oldsource_geocoded.csv-luigi-tmp-1647603946' mode='wb' encoding='UTF-8'>

代码在这里（我删除了业务模型部分以使其更短）：

import pandas as pd
import geopandas as gp
from geopandas.tools import sjoin
from dask import dataframe as dd
from shapely.geometry import Point
from os import path
import luigi

class geocode_tweets(luigi.Task):
    boundaries = _load_geoboundaries()
    nyc = boundaries[0].unary_union

    def requires(self):
        return []

    def output(self):
        self.path = '../data/2017_04_11_oldsource_geocoded.csv'
        return luigi.LocalTarget(self.path)

    def run(self):
        df = dd.read_csv(path.join(data_dir, '2017_03_22_oldsource.csv'))
        df['geometry'] = df.apply(_get_point, axis=1)
        meta = _form_meta(df)

        S = df.map_partitions(
            distributed_sjoin, boundaries=self.boundaries,
            nyc_border=self.nyc, meta=meta).drop('geometry', axis=1)

        f = self.output().open('w')
        S.to_csv(f)
        f.close()

看起来问题出在输出部分

据我了解，问题是 dask 不喜欢将 Luigi 文件对象替换为字符串。

Answer 1

Dask 定义了 DataFrame.to_csv(filename, **kwargs) 并且您发送给它的是文件而不是文件名。将最后三行替换为：

S.to_csv(self.output().path)

dask + luigi: 提高 ValueError('url type not understood: %s' % urlpath)

dask + luigi: raise ValueError('url type not understood: %s' % urlpath)

luigi

dask