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

Question

I am trying to merge dask with luigi, and while business logic works fine by itself, code starts throwing errors when I run a Luigi task:

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'>

the code is here (I dropped the business model part to make it shorter):

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()

and the problem, it looks like, is in the output part

As far as I understand, problem is that dask does not like Luigi file objects as a substitution to the string.

Answer 1

Dask defines DataFrame.to_csv(filename, **kwargs) and you are sending it a file instead of a filename. Replace those last three lines with:

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