code: def _persist(source, path, encoder=None):
          
          import posixpath
          from dask.bytes import open_files
          import dask
          import pickle
          import json
          from intake.source.textfiles import TextFilesSource
          encoder = {None: str, 'str': str, 'json': json.dumps,
                     'pickle': pickle.dumps}[encoder]
          try:
              b = source.to_dask()
          except NotImplementedError:
              import dask.bag as db
              b = db.from_sequence(source.read(), npartitions=1)
          files = open_files(posixpath.join(path, 'part.*'), mode='wt',
                             num=b.npartitions)
          dwrite = dask.delayed(write_file)
          out = [dwrite(part, f, encoder)
                 for part, f in zip(b.to_delayed(), files)]
          dask.compute(out)
          s = TextFilesSource(posixpath.join(path, 'part.*'))
          return s
docs: Save list to files using encoding
  
          encoder : None or one of str|json|pickle
              None is equivalent to str