code: func (d *Dataset) Output(f func(io.Reader) error) *Dataset {
  	step := d.Flow.AddAllToOneStep(d, nil)
  	step.IsOnDriverSide = true
  	step.Name = "Output"
  	step.Function = func(readers []io.Reader, writers []io.Writer, stat *pb.InstructionStat) error {
  		errChan := make(chan error, len(readers))
  		for i, reader := range readers {
  			go func(i int, reader io.Reader) {
  				errChan <- f(reader)
  			}(i, reader)
  		}
  		for range readers {
  			err := <-errChan
  			if err != nil {
  				fmt.Fprintf(os.Stderr, "Failed to process output: %v\n", err)
  				return err
  			}
  		}
  		return nil
  	}
  	return d
  }
docs: // Output concurrently collects outputs from previous step to the driver.