streamDataProcessing.js

import { createReadStream } from 'fs';
import { pipeline, PassThrough } from 'stream';
import parse from 'csv-parse';

const csvParser = parse({ columns: true });

const crimesDataFile = createReadStream('./data/london_crime_by_lsoa.csv');

class AbstractDataAggregator {
  constructor() {
    if (this.constructor === AbstractDataAggregator) {
      throw new Error('Class can\'t be instantiated.');
    }
    this.data = {};
    this.answer = {};
  }

  get processedData() {
    return this.data;  
  }

  processData() {
    throw new Error("Method 'processData()' must be implemented.");
  }

  answerQuestion() {
    throw new Error("Method 'answerQuestion()' must be implemented.");
  }
}

class CommonCrimesAggregator extends AbstractDataAggregator {
  processData(record) {
    const { borough, minor_category, value } = record;
    if (!this.data[borough] && Number(value)) {
      this.data[borough] = { [minor_category]: Number(value) };
    } else if (this.data[borough]) {
      if(!this.data[borough][minor_category]) {
        this.data[borough][minor_category] = Number(value);
      } else {
        this.data[borough][minor_category] += Number(value);
      }
    }
  }

  answerQuestion() {
    let maxNum = 0;
    let minNum = 0;
    let maxCrime = null;
    let minCrime = null;

    const minMaxCrimes = {};

    Object.entries(this.data).forEach(([key, value]) => {
      Object.entries(value).forEach(([crime, number], idx, arr) => {
        if(arr[idx + 1]) {
          const [crimeType, crimeCount] = arr[idx + 1];
          if(!minNum) minNum = number;

          if (crimeCount > maxNum) {
            maxNum = crimeCount;
            maxCrime = crimeType;
          }
          if (minNum && crimeCount < minNum) {
            minNum = crimeCount;
            minCrime = crimeType;
          }
        }
      });
      minMaxCrimes[key] = { [maxCrime]: maxNum, [minCrime]: minNum };
      maxNum = 0;
      minNum = 0;
      maxCrime = null;
      minCrime = null;
    });

    this.answer = minMaxCrimes

    return this.answer;
  }
}

class YearlyCrimesAggregator extends AbstractDataAggregator {
  processData(record) {
    const { year, value } = record;
    if (!this.data[year]) {
      this.data[year] = Number(value);
    } else if (this.data[year] && Number(value)) {
      this.data[year] += Number(value);
    }
  }

  answerQuestion() {
    return this.data;
  }
}

class DangerousAreasAggregator extends AbstractDataAggregator {
  processData(record) {
    const { borough, value } = record;
    if (!this.data[borough] && Number(value)) {
      this.data[borough] = Number(value);
    } else if (this.data[borough] && Number(value)) {
      this.data[borough] += Number(value);
    }
  }

  answerQuestion() {
    return this.answer = Object.entries(this.data)
    .sort(([,a], [,b]) => b - a)
    .reduce((acc, [key, val]) => ({ ...acc, [key]: val }), {});
  }
}

const dataAggregationObj = {
  commonCrimes: new CommonCrimesAggregator(),
  yearlyCrimes: new YearlyCrimesAggregator(),
  dangerousAreas: new DangerousAreasAggregator(),
};

const createDataAggregator = aggregationType => {
  const monitor = new PassThrough({ objectMode: true });

  monitor.on('data', chunk => dataAggregationObj[aggregationType].processData(chunk));

  return monitor;
}

pipeline(
  crimesDataFile,
  csvParser,
  createDataAggregator('commonCrimes'),
  createDataAggregator('yearlyCrimes'),
  createDataAggregator('dangerousAreas'),
  err => {
    if (err) {
      console.error('pipeline failed: ', err);
      process.exit(1);
    }
    const { commonCrimes, yearlyCrimes, dangerousAreas } = dataAggregationObj;
    console.log('most common and least common crimes:', commonCrimes.answerQuestion());
    console.log('crimes over the years: ', yearlyCrimes.answerQuestion());
    console.log('most dangerous areas of London: ', dangerousAreas.answerQuestion());
  },
);