lambda and cloudwatch trigger

f1_index.js

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

const uri = 'mongodb://localhost:27017/mydb';
const dbName = 'mydb';
const collectionName = 'mycollection';
const elasticUrl = 'http://localhost:9200';
const indexName = 'myindex';

exports.handler = async function(event, context) {
  const client = new MongoClient(uri, { useNewUrlParser: true, useUnifiedTopology: true });
  const elasticClient = new Client({ node: elasticUrl });

  try {
    await client.connect();
    const collection = client.db(dbName).collection(collectionName);
    const data = await collection.find({}).toArray();
    const transformedData = transformData(data);
    const result = await elasticClient.bulk({ body: transformedData });
    console.log(`Sent ${result.body.items.length} documents to Elasticsearch`);
  } catch (error) {
    console.error(`Failed to transfer data: ${error.message}`);
    throw error;
  } finally {
    await client.close();
    await elasticClient.close();
  }
};

function transformData(data) {
  return data.map(doc => ({
    index: {
      _index: indexName,
      _type: '_doc',
      _id: doc._id.toString()
    }
  }), {
    ...doc,
    
  });
}

f2_batch.js

const maxRetries = 3;
const retryDelay = 1000; // 1 second

async function sendBatch(batch) {
  let retries = 0;
  while (retries < maxRetries) {
    try {
      const transformedData = transformData(batch);
      const result = await elasticClient.bulk({ body: transformedData });
      console.log(`Sent ${result.body.items.length} documents to Elasticsearch`);
      return;
    } catch (error) {
      console.error(`Failed to send batch: ${error.message}`);
      retries++;
      await new Promise(resolve => setTimeout(resolve, retryDelay * retries));
    }
  }
  console.error(`Failed to send batch after ${maxRetries} retries`);
}

f3_sendByPageSize.js

const pageSize = 1000;
let page = 0;
let data = [];
let hasMore = true;

while (hasMore) {
  data = await collection.find({})
    .skip(page * pageSize)
    .limit(pageSize)
    .toArray();
  hasMore = data.length === pageSize;
  const transformedData = transformData(data);
  const result = await elasticClient.bulk({ body: transformedData });
  console.log(`Sent ${result.body.items.length} documents to Elasticsearch`);
  page++;
}

f4_createWithCronJob.js

const cron = require('node-cron');

cron.schedule('* * * * *', async () => {
  await mongoClient.connect();
  const db = mongoClient.db('your-db-name');
  const collection = db.collection('your-collection-name');
  const data = await collection.find({}).toArray();
  const transformedData = transformData(data);
  const result = await elasticClient.bulk({ body: transformedData });
  console.log(`Sent ${result.body.items.length} documents to Elasticsearch`);
  await mongoClient.close();
});

You can create a lambda function for trigger with CloudWatch each min.

aws events put-rule --name "TransferDataRule" --schedule-expression "rate(1 minute)"
this is cli code for creating cloud watch trigger. You can also create on AWS console