streamってこんなかんじ？

datapatch_stream_sample.coffee

jsforce = require 'jsforce'
Promise = require 'bluebird'
streamify = require 'stream-array'
csv = require 'csv'
fs = require 'fs'
program = require 'commander'

program
.option '-d, --deploy', 'deploy'
.parse process.argv

require('dotenv').load()
username = process.env.SALESFORCE_USERNAME
password = process.env.SALESFORCE_PASSWORD
host = process.env.SALESFORCE_HOST || 'login.salesforce.com'
host = "https://#{host}" unless host.match(/^https:\/\//)

connection = new jsforce.Connection loginUrl: host
connection.login username, password
.then ->
  task = ->
    if program.deploy
      connection.sobject('Contact').updateBulk()
      .on 'response', (results) ->
        streamify results.map (e) ->
          id: e.id, success: String(e.success), errors: e.errors.join('\n')
        .pipe csv.stringify(header: true)
        .pipe fs.createWriteStream "./results-#{new Date().toISOString()}.csv"

        error_count = results.filter((e) -> !e.success).length
        console.error "#{error_count} errors" if error_count > 0
      .on 'error', ->
        console.error arguments
      .stream()
    else fs.createWriteStream "./dryrun-#{new Date().toISOString()}.csv"

  connection.bulk.query('select Id, Name, LastName from Contact')
  .on 'error', ->
    console.error arguments
  .stream()
  .pipe csv.parse(columns: true)
  .pipe csv.transform (e) ->
    # データを加工してreadonly項目を除去
    e.LastName = e.LastName + e.Name
    delete e.Name
    return e
  .pipe csv.stringify(header: true)
  .pipe task()
.fail ->
  console.error arguments
  

package.json

{
  "name": "datapatch_stream_sample",
  "dependencies": {
    "bluebird": "^2.4.2",
    "commander": "^2.5.1",
    "csv": "^0.4.1",
    "dotenv": "^0.4.0",
    "jsforce": "^1.3.1",
    "stream-array": "^1.0.1"
  }
}