A more efficient way to stream data from a Cassandra to Postgres Database

[realnsleo]

Hello @vitaly-t,
First off I would like to thank you for the amazing work.
I have a pretty simple use case. I have a Cassandra setup with tables containing millions of records. I wish to migrate that data to a PostgreSQL database. I am using the code below to achieve this purpose (using cassandra-driver and pg-promise). I have stripped off a lot of boilerplate to show the main part of the application.

const cassandra = require('cassandra-driver');
const promise = require('bluebird');
const fs = require('fs');

const initOptions = {
  promiseLib: promise
};

const pgp = require('pg-promise')(initOptions);

async function run(fromDate, toDate) {

  const cassandraClient = new cassandra.Client({ options });

  const db = pgp({
    host: process.env.PG_HOST,
    database: process.env.PG_DATABASE,
    user: process.env.PG_USER,
    password: process.env.PG_PASSWORD,
    port: process.env.PG_PORT,
    allowExitOnIdle: true,
    ssl: {
      rejectUnauthorized: false,
      ca: fs.readFileSync("/path/to/my/ca/file").toString(),
    }
  });

  await cassandraClient.connect();

  let records = [];

  const cs = new pgp.helpers.ColumnSet(['id', 'reference_number', 'amount', 'created_at'], { table: 'transactions' });

  // A query like this returns approximately 1 million rows
  cassandraClient.stream(
    `SELECT * FROM transaction_by_id WHERE time >= ? AND time <= ? ALLOW FILTERING`,
    [fromDate, toDate], {
    prepare: true,
    readTimeout: 20000
  })

    // The rows from cassandra are streamed here
    .on('readable', async function () {
      let row;

      while (row = this.read()) {
        records.push(row)
      }
    })

    // Cassandra has successfully fetched all tows
    .on('end', async function () {
      const query = pgp.helpers.insert(records, cs);
      await db.none(query);
    })

    .on('error', async function (err) {
      throw new Error(err.message);
    });

  run(fromDate, toDate);

}

The problem is whenever I run this query I get errors happening about 5 minutes later. The errors range fro timeouts, socket shutdown and connection closed issues. An example of the error is below

Error: Connection terminated unexpectedly
    at Connection.<anonymous> (/home/ubuntu/scripts/report-dumper/node_modules/pg/lib/client.js:132:73)
    at Object.onceWrapper (node:events:639:28)
    at Connection.emit (node:events:520:28)
    at TLSSocket.<anonymous> (/home/ubuntu/scripts/report-dumper/node_modules/pg/lib/connection.js:107:12)
    at TLSSocket.emit (node:events:532:35)
    at endReadableNT (node:internal/streams/readable:1346:12)
    at processTicksAndRejections (node:internal/process/task_queues:83:21)

I wanted to know if there is a much better way of achieving this goal? I read something here: LINK and I was wondering how I can stream the data into the database. I appreciate any assistance or general guidance. Thanks.

[vitaly-t]

Are you using the latest version of pg-promise?