Share via

Gebruik Node.js om verbinding te maken en SQL-opdrachten uit te voeren in Azure Cosmos DB for PostgreSQL

  • Artikel

VAN TOEPASSING OP: Azure Cosmos DB for PostgreSQL (mogelijk gemaakt door de Citus-databaseextensie voor PostgreSQL)

In deze quickstart ziet u hoe u Node.js code gebruikt om verbinding te maken met een cluster en hoe u SQL-instructies gebruikt om een tabel te maken. Vervolgens gaat u gegevens in de database invoegen, opvragen, bijwerken en verwijderen. Bij de stappen in dit artikel wordt ervan uitgegaan dat u bekend bent met Node.js ontwikkeling en dat u geen ervaring hebt met het werken met Azure Cosmos DB for PostgreSQL.

PostgreSQL-bibliotheek installeren

Voor de codevoorbeelden in dit artikel is vereist dat de pg-bibliotheek een interface heeft met de PostgreSQL-server. U moet pg installeren met uw taalpakketbeheer (zoals npm).

Verbinding maken, een tabel maken en gegevens invoegen

De algemene verbindingsmodule maken

Tip

In de onderstaande voorbeeldcode wordt een verbindingsgroep gebruikt om verbindingen met PostgreSQL te maken en te beheren. Groepsgewijze verbindingen aan de toepassingszijde wordt sterk aanbevolen omdat:

  • Het zorgt ervoor dat de toepassing niet te veel verbindingen met de database genereert en zo wordt voorkomen dat de verbindingslimieten worden overschreden.
  • Het kan helpen de prestaties drastisch te verbeteren, zowel latentie als doorvoer. Het PostgreSQL-serverproces moet fork zijn om elke nieuwe verbinding af te handelen. Als u een verbinding opnieuw gebruikt, wordt die overhead vermeden.

Maak een map met de naam db en maak in deze map een citus.js-bestand dat de volgende algemene verbindingscode bevat. Vervang in deze code cluster> door <uw clusternaam en <wachtwoord> door uw beheerderswachtwoord.

/**
* file: db/citus.js
*/

const { Pool } = require('pg');

const pool = new Pool({
  max: 300,
  connectionTimeoutMillis: 5000,

  host: 'c-<cluster>.<uniqueID>.postgres.cosmos.azure.com',
  port: 5432,
  user: 'citus',
  password: '<password>',
  database: 'citus',
  ssl: true,
});

module.exports = {
  pool,
};

Een tabel maken

Gebruik de volgende code om verbinding te maken en de gegevens te laden met behulp van de SQL-instructies CREATE TABLE en INSERT INTO. Met de code wordt een nieuwe pharmacy tabel gemaakt en worden enkele voorbeeldgegevens ingevoegd.

/**
* file: create.js
*/

const { pool } = require('./db/citus');

async function queryDatabase() {
  const queryString = `
    DROP TABLE IF EXISTS pharmacy;
    CREATE TABLE pharmacy (pharmacy_id integer,pharmacy_name text,city text,state text,zip_code integer);
    INSERT INTO pharmacy (pharmacy_id,pharmacy_name,city,state,zip_code) VALUES (0,'Target','Sunnyvale','California',94001);
    INSERT INTO pharmacy (pharmacy_id,pharmacy_name,city,state,zip_code) VALUES (1,'CVS','San Francisco','California',94002);
    INSERT INTO pharmacy (pharmacy_id,pharmacy_name,city,state,zip_code) VALUES (2,'Walgreens','San Diego','California',94003);
    CREATE INDEX idx_pharmacy_id ON pharmacy(pharmacy_id);
  `;

  try {
    /* Real application code would probably request a dedicated client with
       pool.connect() and run multiple queries with the client. In this
       example, you're running only one query, so you use the pool.query()
       helper method to run it on the first available idle client.
    */

    await pool.query(queryString);
    console.log('Created the Pharmacy table and inserted rows.');
  } catch (err) {
    console.log(err.stack);
  } finally {
    pool.end();
  }
}

queryDatabase();

Tabellen distribueren

Azure Cosmos DB for PostgreSQL biedt u de superkracht van het distribueren van tabellen over meerdere knooppunten voor schaalbaarheid. Met de onderstaande opdracht kunt u een tabel distribueren. Meer informatie over create_distributed_table en de distributiekolom vindt u hier.

Notitie

Als u tabellen distribueert, kunnen deze worden vergroot over alle werkknooppunten die aan het cluster zijn toegevoegd.

Gebruik de volgende code om verbinding te maken met de database en de tabel te distribueren.

/**
* file: distribute-table.js
*/

const { pool } = require('./db/citus');

async function queryDatabase() {
  const queryString = `
    SELECT create_distributed_table('pharmacy', 'pharmacy_id');
  `;

  try {
    await pool.query(queryString);
    console.log('Distributed pharmacy table.');
  } catch (err) {
    console.log(err.stack);
  } finally {
    pool.end();
  }
}

queryDatabase();

Gegevens lezen

Gebruik de volgende code om verbinding te maken en de gegevens te lezen met behulp van de SQL-instructie SELECT.

/**
* file: read.js
*/

const { pool } = require('./db/citus');

async function queryDatabase() {
  const queryString = `
    SELECT * FROM pharmacy;
  `;

  try {
    const res = await pool.query(queryString);
    console.log(res.rows);
  } catch (err) {
    console.log(err.stack);
  } finally {
    pool.end();
  }
}

queryDatabase();

Gegevens bijwerken

Gebruik de volgende code om verbinding te maken en de gegevens bij te werken met behulp van de SQL-instructie UPDATE.

/**
* file: update.js
*/

const { pool } = require('./db/citus');

async function queryDatabase() {
  const queryString = `
    UPDATE pharmacy SET city = 'Long Beach'
    WHERE pharmacy_id = 1;
  `;

  try {
    const result = await pool.query(queryString);
    console.log('Update completed.');
    console.log(`Rows affected: ${result.rowCount}`);
  } catch (err) {
    console.log(err.stack);
  } finally {
    pool.end();
  }
}

queryDatabase();

Gegevens verwijderen

Gebruik de volgende code om verbinding te maken en de gegevens te verwijderen met behulp van de SQL-instructie DELETE.

/**
* file: delete.js
*/

const { pool } = require('./db/citus');

async function queryDatabase() {
  const queryString = `
    DELETE FROM pharmacy
    WHERE pharmacy_name = 'Target';
  `;

  try {
    const result = await pool.query(queryString);
    console.log('Delete completed.');
    console.log(`Rows affected: ${result.rowCount}`);
  } catch (err) {
    console.log(err.stack);
  } finally {
    pool.end();
  }
}

queryDatabase();

COPY-opdracht voor snelle opname

De opdracht COPY kan een enorme doorvoer opleveren tijdens het opnemen van gegevens in Azure Cosmos DB for PostgreSQL. De opdracht COPY kan gegevens opnemen in bestanden of uit microbatches met gegevens in het geheugen voor realtimeopname.

COPY-opdracht voor het laden van gegevens uit een bestand

Met de volgende code worden gegevens uit een CSV-bestand gekopieerd naar een databasetabel. De code vereist het pakket pg-copy-streams en het bestand pharmacies.csv.

/**
* file: copycsv.js
*/

const inputFile = require('path').join(__dirname, '/pharmacies.csv');
const fileStream = require('fs').createReadStream(inputFile);
const copyFrom = require('pg-copy-streams').from;
const { pool } = require('./db/citus');

async function importCsvDatabase() {
  return new Promise((resolve, reject) => {
    const queryString = `
      COPY pharmacy FROM STDIN WITH (FORMAT CSV, HEADER true, NULL '');
    `;

    fileStream.on('error', reject);

    pool
      .connect()
      .then(client => {
        const stream = client
          .query(copyFrom(queryString))
          .on('error', reject)
          .on('end', () => {
            reject(new Error('Connection closed!'));
          })
          .on('finish', () => {
            client.release();
            resolve();
          });

        fileStream.pipe(stream);
      })
      .catch(err => {
        reject(new Error(err));
      });
  });
}

(async () => {
  console.log('Copying from CSV...');
  await importCsvDatabase();
  await pool.end();
  console.log('Inserted csv successfully');
})();

COPY-opdracht voor het laden van gegevens in het geheugen

Met de volgende code worden gegevens in het geheugen gekopieerd naar een tabel. Voor de code is het pakket through2 vereist, waarmee pipe chaining mogelijk is.

/**
 * file: copyinmemory.js
 */

const through2 = require('through2');
const copyFrom = require('pg-copy-streams').from;
const { pool } = require('./db/citus');

async function importInMemoryDatabase() {
  return new Promise((resolve, reject) => {
    pool
      .connect()
      .then(client => {
        const stream = client
          .query(copyFrom('COPY pharmacy FROM STDIN'))
          .on('error', reject)
          .on('end', () => {
            reject(new Error('Connection closed!'));
          })
          .on('finish', () => {
            client.release();
            resolve();
          });

        const internDataset = [
          ['100', 'Target', 'Sunnyvale', 'California', '94001'],
          ['101', 'CVS', 'San Francisco', 'California', '94002'],
        ];

        let started = false;
        const internStream = through2.obj((arr, _enc, cb) => {
          const rowText = (started ? '\n' : '') + arr.join('\t');
          started = true;
          cb(null, rowText);
        });

        internStream.on('error', reject).pipe(stream);

        internDataset.forEach((record) => {
          internStream.write(record);
        });

        internStream.end();
      })
      .catch(err => {
        reject(new Error(err));
      });
  });
}
(async () => {
  await importInMemoryDatabase();
  await pool.end();
  console.log('Inserted inmemory data successfully.');
})();

App opnieuw proberen voor databaseaanvraagfouten

Het is soms mogelijk dat databaseaanvragen van uw toepassing mislukken. Dergelijke problemen kunnen zich voordoen in verschillende scenario's, zoals netwerkfouten tussen app en database, onjuist wachtwoord, enzovoort. Sommige problemen kunnen tijdelijk zijn en zichzelf binnen enkele seconden tot minuten oplossen. U kunt logica voor opnieuw proberen in uw app configureren om de tijdelijke fouten te verhelpen.

Het configureren van logica voor opnieuw proberen in uw app helpt de eindgebruikerservaring te verbeteren. In foutscenario's wachten gebruikers alleen iets langer totdat de toepassing aanvragen verwerkt, in plaats van fouten te ervaren.

In het onderstaande voorbeeld ziet u hoe u logica voor opnieuw proberen implementeert in uw app. Het voorbeeldcodefragment probeert elke 60 seconden (maximaal vijf keer) een databaseaanvraag uit te voeren totdat dit lukt. Het aantal en de frequentie van nieuwe pogingen kunnen worden geconfigureerd op basis van de behoeften van uw toepassing.

Vervang in deze code cluster> door <uw clusternaam en <wachtwoord> door uw beheerderswachtwoord.

const { Pool } = require('pg');
const { sleep } = require('sleep');

const pool = new Pool({
  host: 'c-<cluster>.<uniqueID>.postgres.cosmos.azure.com',
  port: 5432,
  user: 'citus',
  password: '<password>',
  database: 'citus',
  ssl: true,
  connectionTimeoutMillis: 0,
  idleTimeoutMillis: 0,
  min: 10,
  max: 20,
});

(async function() {
  res = await executeRetry('select nonexistent_thing;',5);
  console.log(res);
  process.exit(res ? 0 : 1);
})();

async function executeRetry(sql,retryCount)
{
  for (let i = 0; i < retryCount; i++) {
    try {
      result = await pool.query(sql)
      return result;
    } catch (err) {
      console.log(err.message);
      sleep(60);
    }
  }

  // didn't succeed after all the tries
  return null;
}

Volgende stappen