Check against a schema.

chkagainstschema.js

var check = require('check-types')
  , fs = require('fs')
  , csv = require('fast-csv')
  , _ = require('lodash')
  , stream = fs.createReadStream("Ptax_Property_Data_5_Yr.csv");

function valueToType(v) {
  if (isNaN(v)) {
    return v;
  } else {
    return eval(v);
  }
}

function checkColumn(schema, column, value) {
  if (schema[column].type == 'ignore'){
    return [];
  } else if (!check[schema[column].type](valueToType(value))) {
    return ['Column must be a ' + schema[column].type + "."];
  } else {
    return [];
  }
}

/* Schema */
var columns = {

  STATE_TAX_MAP: {type: 'string'}, 
  TAX_YEAR: {type: 'number'},
  BILL_NBR: {type: 'number'},
  OWNER_NAME_1: {type: 'string'},
  OWNER_NAME_2: {type: 'string'},
  OWNER_ADDRESS_1: {type: 'string'},
  OWNER_ADDRESS_2: {type: 'string'},
  PROPERTY_ADDRESS: {type: 'string'},
  TAX_AMOUNT_DUE: {type: 'number'},
  ASSESSED_TAXES: {type: 'number'},
  ASSESSED_VALUE: {type: 'number'},
  TAX_AMT_PAID: {type: 'number'},
  DELQ_TAX_AMT_PAID: {type: 'number'},
  AMOUNT_PAID: {type: 'number'},
  DATE_TAXES_PAID: {type: 'string'},
  TAX_CHANGE: {type: 'string'},
  TAX_REFUND: {type: 'number'},
  PENALTY_AMT_PAID: {type: 'number'},
  PENALTY_REFUND: {type: 'number'},
  OVERPAYMENT_AMT_PAID: {type: 'number'},
  OVERPAYMENT_REFUND: {type: 'number'},
  DISCOUNT_AMT_CREDITED: {type: 'number'},
  DISCOUNT_REFUND: {type: 'number'},
  COMM_FEE_REFUND: {type: 'number'},
  COMMISSION_FEE: {type: 'number'},
  DELQ_COMMISSION_FEE_PAID: {type: 'number'},
  INTEREST_PENALTY: {type: 'number'},
  DELQ_PENALTY_AMT_PAID: {type: 'number'},
  COURT_COST: {type: 'number'},
  COURT_COST_PAID: {type: 'number'},
  ACV_CREDITED: {type: 'number'},
  TAXES_DUE_AT_ROLLOVER: {type: 'number'},
  MTG_CPNY_CODE: {type: 'string'},
  WQ_FEE_ASSESSED: {type: 'number'},
  WQ_INTEREST_COMPOUNDED: {type: 'number'},
  WQ_FEE_TOTAL_PAID: {type: 'number'},
  WQ_INT_TOTAL_PAID: {type: 'number'},
  WQ_BALANCE_DUE: {type: 'number'},
  WQ_FEE_TOTAL_REFUND: {type: 'number'},
  WQ_INT_TOTAL_REFUND: {type: 'number'},
  BANKRUPTCY_IND: {type: 'string'},
  DELINQUENT_IND: {type: 'string'},
  TYPE_OF_TAX_IND: {type: 'ignore'},
  CREATE_DATE: {type: 'string'},
  MAINT_DATE: {type: 'string'},
};

var ctr = 1;

console.log("[");
var csvStream = csv({objectMode: true, headers: true})
  .on("record", function(data){
    var dta = _.chain(_.keys(data))
      .map(function (d) { return {column: d, value: data[d], errors: checkColumn(columns, d, data[d])};})
      .filter(function (d) { return d.errors.length > 0; })
      .value();
    console.log('%j,', {row: ctr, data: dta});
    ctr += 1;
 })
 .on("end", function(){
     console.log("]");
 });

stream.pipe(csvStream);