How to use the pelias-model.createDocumentMapperStream function in pelias-model

var proxiedPbf = proxyquire('../stream/multiple_pbfs', {'pelias-config' : fakePeliasConfig});

streams = proxyquire('../stream/importPipeline', {'./multiple_pbfs': proxiedPbf});

var expectedPath = path.resolve(__dirname) + '/fixtures/combined_vancouver_queens.json';

var results = [];

streams.pbfParser()
  .pipe( streams.docConstructor() )
  .pipe( streams.tagMapper() )
  .pipe( streams.addressExtractor() )
  .pipe( streams.categoryMapper( streams.config.categoryDefaults ) )
  .pipe( streams.addendumMapper() )
  .pipe( model.createDocumentMapperStream() )
  .pipe( sink.obj(function (doc) {
    results.push(doc);
  }) )
  .on('error', function error(err) {
    console.error('YIKES! Test failed with the following error: ', err.message);
    process.exit();
  })
  .on('finish', function assert(){

    var actual = results;
    var expected = JSON.parse( fs.readFileSync( expectedPath, { encoding: 'utf8' } ) );

    actual = _.sortBy(actual, ['_id']);
    expected = _.sortBy(expected, ['_id']);

    fs.writeFileSync('actual_output.json', JSON.stringify(actual, null, 2));

function fullImport(wof_record_stream, document_generator, destination_pipe, callback) {
  wof_record_stream
    .pipe(document_generator)
    .pipe(blacklistStream())
    .pipe(peliasModel.createDocumentMapperStream())
    .pipe(destination_pipe)
    .on('finish', callback);

}