Parse input CSV and stream to Redis Set

redis-set-loader-node-sample-input.csv

redis-set-loader-node.js

const redis = require("redis");
const fs = require('fs');
const parse = require('csv-parse');
const { Writable } = require('stream');

const argv = require('yargs').options({
    input: { type: 'string', describe: 'CSV file with input to be loaded', demandOption: true },
    conn: { default: 'redis://127.0.0.1:6379', type: 'string', describe: 'Redis connection string', demandOption: true },
    set: { default: 'my-set', type: 'string', describe: 'Name of the set to be loaded', demandOption: true },
}).usage('Usage: $0 --input=[FILE] --conn=[CONN-STR] --set=[SET-NAME]').argv;

/* * * Testing this script locally: 
 * docker run -d --name jv-redis -p 6379:6379 redis
 * node redis-set-loader.js --input test.csv --conn redis://127.0.0.1:6379 --set some-set
 * docker exec -it jv-redis redis-cli SMEMBERS some-set
 * * */

class RedisStream extends Writable {

    constructor(conn, set) {
        super({ objectMode: true });
        this.client = redis.createClient(conn);
        this.set = set;
    }

    _write(row, encoding, callback) {
        const item = row.toString();
        this.client.sadd(this.set, item, err => callback(err));
    }
}

let main = () => {
    const input = argv.input;
    const conn = argv.conn;
    const set = argv.set;

    const fileIn = fs.createReadStream(input);
    const redisOut = new RedisStream(conn, set);
    const parser = parse();

    fileIn
        .pipe(parser)
        .pipe(redisOut)
        .on('finish', () => {
            console.log('All data streamed.');
            process.exit(0);
        });
};

main();