fix(datasets): Use lazy .sink_* instead of collecting data

fix(datasets): Use lazy .sink_* instead of collecting data #1678