基于NodeJs+PostgreSQL的应用：如何向ElasticSearch添加数据并同步？

Hey there! Let's break down how to add data to Elasticsearch and keep your PostgreSQL data in sync with it for your Node.js app. I'll walk you through practical, actionable steps and examples that you can drop into your code right away.

First, you'll need the official Elasticsearch client for Node.js—it's the most reliable way to interact with ES from your app. Install it via npm:

npm install @elastic/elasticsearch

Next, set up your client connection. Make sure to replace the node URL with your actual Elasticsearch instance (and add auth credentials if you're using secured ES):

const { Client } = require('@elastic/elasticsearch');

const esClient = new Client({
  node: 'http://localhost:9200',
  // auth: { username: 'elastic', password: 'your-secure-password' } // Uncomment if needed
});

Adding a Single Document #

Indexing a single record is straightforward. Here's a reusable function to add a document to your target index (swap products with your index name):

async function indexSingleDocument(doc) {
  try {
    const response = await esClient.index({
      index: 'products',
      id: doc.id.toString(), // Optional: Let ES auto-generate an ID if you omit this
      document: doc
    });
    console.log(`Document ${doc.id} indexed successfully:`, response.result);
  } catch (err) {
    console.error(`Failed to index document ${doc.id}:`, err.message);
  }
}

// Usage example
indexSingleDocument({ id: 1, name: 'Wireless Headphones', price: 89.99, category: 'Electronics' });

Bulk Inserting Multiple Documents #

For large datasets, bulk operations are way more efficient than indexing one-by-one. Here's how to batch multiple documents:

async function bulkIndexDocuments(docs) {
  // Format operations for ES bulk API: each document needs an "index" action followed by the data
  const operations = docs.flatMap(doc => [
    { index: { _index: 'products', _id: doc.id.toString() } },
    doc
  ]);

  try {
    const response = await esClient.bulk({ operations });
    // Check for failed operations if needed
    if (response.errors) {
      console.warn('Some documents failed to index:', response.items.filter(item => item.index.error));
    } else {
      console.log(`Successfully indexed ${docs.length} documents`);
    }
  } catch (err) {
    console.error('Bulk index failed:', err.message);
  }
}

// Usage example
bulkIndexDocuments([
  { id: 2, name: 'Smart Light Bulb', price: 19.99, category: 'Home' },
  { id: 3, name: 'Portable Charger', price: 29.99, category: 'Electronics' }
]);

Now, let's cover the main approaches to keep your PostgreSQL data in lockstep with Elasticsearch. The right choice depends on your app's scale, traffic, and reliability needs.

1. Application-Level Sync (Quick & Simple for Small Apps) #

If your app has low-to-moderate traffic and all data changes go through your Node.js code, this is the easiest approach. Just sync to ES right after you write to PostgreSQL.

Example: Sync After Saving to PostgreSQL #

Assuming you're using the pg library for PostgreSQL:

const { Pool } = require('pg');
const pgPool = new Pool({
  user: 'your-db-user',
  host: 'localhost',
  database: 'your-db-name',
  password: 'your-db-password',
  port: 5432
});

async function saveProductToPostgres(product) {
  const { rows } = await pgPool.query(
    'INSERT INTO products (name, price, category) VALUES ($1, $2, $3) RETURNING *',
    [product.name, product.price, product.category]
  );
  return rows[0];
}

// Wrap the save with ES sync
async function createProduct(product) {
  const savedProduct = await saveProductToPostgres(product);
  await indexSingleDocument(savedProduct); // Reuse our index function
  return savedProduct;
}

// Usage
createProduct({ name: 'Bluetooth Speaker', price: 49.99, category: 'Electronics' });

For updates and deletes, do the same: after updating PostgreSQL, call esClient.update(); after deleting, call esClient.delete().

Pros: Zero extra infrastructure, easy to implement.
Cons: Doesn't catch changes made directly to PostgreSQL (like manual SQL queries), and if your app crashes mid-sync, data will be out of sync.

2. Change Data Capture (CDC) with Debezium (Most Robust for Production) #

CDC is the industry standard for reliable, real-time database sync. It captures every change (INSERT/UPDATE/DELETE) directly from PostgreSQL's write-ahead log (WAL), so it catches everything—even changes outside your app.

Debezium is a popular open-source CDC tool that integrates seamlessly with PostgreSQL and Elasticsearch. Here's the high-level workflow:

• Set up a Debezium connector for PostgreSQL, configured to monitor your target tables.
• Debezium takes an initial snapshot of your existing PostgreSQL data and syncs it to Elasticsearch.
• After that, it sends real-time change events directly to Elasticsearch (or uses Kafka as an intermediary for scalability).

Pros: Catches all database changes, resilient to failures, real-time sync, no impact on your app code.
Cons: Requires setting up extra infrastructure (Debezium, optionally Kafka), has a steeper learning curve.

3. PostgreSQL Triggers + Node.js Polling (Middle Ground) #

If you don't want to set up Debezium but need to catch all database changes, you can use PostgreSQL triggers to log changes to a "changelog" table, then have a Node.js script poll this table and sync to ES.

Step 1: Create a Changelog Table #

CREATE TABLE product_changelog (
  id SERIAL PRIMARY KEY,
  product_id INT REFERENCES products(id),
  operation_type VARCHAR(10) CHECK (operation_type IN ('INSERT', 'UPDATE', 'DELETE')),
  changed_data JSONB,
  synced BOOLEAN DEFAULT FALSE,
  created_at TIMESTAMP DEFAULT NOW()
);

Step 2: Create a Trigger Function #

This function logs every change to the products table:

CREATE OR REPLACE FUNCTION log_product_change()
RETURNS TRIGGER AS $$
BEGIN
  IF TG_OP = 'INSERT' THEN
    INSERT INTO product_changelog (product_id, operation_type, changed_data)
    VALUES (NEW.id, 'INSERT', to_jsonb(NEW));
    RETURN NEW;
  ELSIF TG_OP = 'UPDATE' THEN
    INSERT INTO product_changelog (product_id, operation_type, changed_data)
    VALUES (NEW.id, 'UPDATE', to_jsonb(NEW));
    RETURN NEW;
  ELSIF TG_OP = 'DELETE' THEN
    INSERT INTO product_changelog (product_id, operation_type, changed_data)
    VALUES (OLD.id, 'DELETE', to_jsonb(OLD));
    RETURN OLD;
  END IF;
END;
$$ LANGUAGE plpgsql;

Step 3: Attach the Trigger to Your Table #

CREATE TRIGGER products_change_trigger
AFTER INSERT OR UPDATE OR DELETE ON products
FOR EACH ROW EXECUTE FUNCTION log_product_change();

Step 4: Node.js Sync Script #

This script polls the changelog table periodically and syncs changes to ES:

async function syncChangelog() {
  try {
    // Fetch unsynced entries
    const { rows } = await pgPool.query(
      'SELECT * FROM product_changelog WHERE synced = FALSE ORDER BY created_at ASC'
    );

    for (const entry of rows) {
      try {
        if (entry.operation_type === 'INSERT' || entry.operation_type === 'UPDATE') {
          await esClient.index({
            index: 'products',
            id: entry.product_id.toString(),
            document: entry.changed_data
          });
        } else if (entry.operation_type === 'DELETE') {
          await esClient.delete({
            index: 'products',
            id: entry.product_id.toString()
          });
        }

        // Mark entry as synced
        await pgPool.query(
          'UPDATE product_changelog SET synced = TRUE WHERE id = $1',
          [entry.id]
        );
      } catch (syncErr) {
        console.error(`Failed to sync changelog entry ${entry.id}:`, syncErr.message);
        // Leave it as unsynced to retry next time
      }
    }
  } catch (queryErr) {
    console.error('Failed to fetch changelog entries:', queryErr.message);
  }
}

// Run sync every 30 seconds (adjust interval based on your needs)
setInterval(syncChangelog, 30000);

Pros: Catches all database changes, no external tools beyond PostgreSQL and Node.js.
Cons: Polling introduces latency, and you need to handle retries for failed syncs.

内容的提问来源于stack exchange，提问作者Nikolay Podolnyy