Сколько запросов может Node-Express запустить одновременно?


У меня есть скрипт, который извлекает 25 000 записей из AWS Athena, которая в основном является реляционной базой данных SQL PrestoDB. Предположим, что я генерирую запрос для каждой из этих записей, что означает, что я должен сделать 25 000 запросов к Athena, а затем, когда данные возвращаются, я должен сделать 25 000 запросов к моему кластеру Redis.

Каким было бы идеальное количество запросов, которые можно сделать за один раз от узла к Афине?

Причина, по которой я спрашиваю, заключается в том, что я пытался сделать это, создавая массив из 25 000 обещаний, а затем вызов Promise.all(promiseArray) на нем, но приложение просто зависло навсегда.

Поэтому я решил вместо этого запускать по 1 за раз и использовать рекурсию, чтобы сращивать первый индекс, а затем передавать оставшиеся записи вызывающей функции после того, как обещание будет выполнено.

Проблема в том, что это занимает целую вечность. Я взял примерно часовой перерыв и вернулся, и там осталось 23 000 записей.

Я попытался погуглить, сколько запросов Node и Athena справлюсь сразу, но ничего не придумал. Я надеюсь, что кто-то может знать что-то об этом и сможет поделиться этим со мной.

Спасибо.

Вот мой код для справки:

В качестве дополнительной заметки я хотел бы сделать по-другому, вместо того чтобы отправлять один запрос за раз, я мог бы отправить 4, 5, 6, 7 или 8 одновременно в зависимости от того, как быстро он будет выполняться.

Кроме того, как кластер узлов повлияет на производительность чего-то вроде вот это?

exports.storeDomainTrends = () => {
return new Promise((resolve, reject)=>{
    athenaClient.execute(`SELECT DISTINCT the_column from "the_db"."the_table"`,
    (err, data) =>  {
        var getAndStoreDomainData = (records) => {
            if(records.length){
                return new promise((resolve, reject) => {
                    var subrecords = records.splice(0, )[0]
                    athenaClient.execute(`
                    SELECT 
                    field,
                    field,
                    field,
                    SUM(field) as field
                    FROM "the_db"."the_table"
                    WHERE the_field IN ('Month') AND the_field = '`+ record.domain_name +`'
                    GROUP BY the_field, the_field, the_field
                    `, (err, domainTrend) => {

                        if(err) {
                            console.log(err)
                            reject(err)
                        }

                        redisClient.set(('Some String' + domainTrend[0].domain_name), JSON.stringify(domainTrend))
                        resolve(domainTrend);
                    })
                })
                .then(res => {
                    getAndStoreDomainData(records);
                })
            }
        }

        getAndStoreDomainData(data);

    })
})

}

2 2

2 ответа:

Используя lib ваш код может выглядеть примерно так:

const Fail = function(reason){this.reason=reason;};
const isFail = x=>(x&&x.constructor)===Fail;
const distinctDomains = () =>
  new Promise(
    (resolve,reject)=>
      athenaClient.execute(
        `SELECT DISTINCT domain_name from "endpoint_dm"."bd_mb3_global_endpoints"`,
        (err,data)=>
          (err)
            ? reject(err)
            : resolve(data)
      )
  );
const domainDetails = domain_name =>
  new Promise(
    (resolve,reject)=>
      athenaClient.execute(
        `SELECT 
        timeframe_end_date,
        agg_type,
        domain_name,
        SUM(endpoint_count) as endpoint_count
        FROM "endpoint_dm"."bd_mb3_global_endpoints"
        WHERE agg_type IN ('Month') AND domain_name = '${domain_name}'
        GROUP BY timeframe_end_date, agg_type, domain_name`,
        (err, domainTrend) =>
            (err)
              ? reject(err)
              : resolve(domainTrend)
        )
  );
const redisSet = keyValue =>
  new Promise(
    (resolve,reject)=>
      redisClient.set(
        keyValue,
        (err,res)=>
          (err)
            ? reject(err)
            : resolve(res)
      )
  );
const process = batchSize => limitFn => resolveValue => domains => 
  Promise.all(
    domains.slice(0,batchSize)
    .map(//map domains to promises
      domain=>
        //maximum 5 active connections
        limitFn(domainName=>domainDetails(domainName))(domain.domain_name)
        .then(
          domainTrend=>
            //the redis client documentation makes no sense whatsoever
            //https://redis.io/commands/set
            //no mention of a callback
            //https://github.com/NodeRedis/node_redis
            //mentions a callback, since we need the return value
            //and best to do it async we will use callback to promise
            redisSet([
              `Endpoint Profiles - Checkin Trend by Domain - Monthly - ${domainTrend[0].domain_name}`,
              JSON.stringify(domainTrend)
            ])
        )
        .then(
          redisReply=>{
            //here is where things get unpredictable, set is documented as 
            //  a synchronous function returning "OK" or a function that
            //  takes a callback but no mention of what that callback recieves
            //  as response, you should try with one or two records to
            //  finish this on reverse engineering because documentation
            //  fails 100% here and can not be relied uppon.
            console.log("bad documentation of redis client... reply is:",redisReply);
            (redisReply==="OK")
              ? domain
              : Promise.reject(`Redis reply not OK:${redisReply}`)
          }
        )
        .catch(//catch failed, save error and domain of failed item
          e=>
            new Fail([e,domain])
        )
    )
  ).then(
    results=>{
      console.log(`got ${batchSize} results`);
      const left = domains.slice(batchSize);
      if(left.length===0){//nothing left
        return resolveValue.conat(results);
      }
      //recursively call process untill done
      return process(batchSize)(limitFn)(resolveValue.concat(results))(left)
    }
  );
const max5 = lib.throttle(5);//max 5 active connections to athena
distinctDomains()//you may want to limit the results to 50 for testing
//you may want to limit batch size to 10 for testing
.then(process(1000)(max5)([]))//we have 25000 domains here
.then(
  results=>{//have 25000 results
    const successes = results.filter(x=>!isFail(x));
    //array of failed items, a failed item has a .reason property
    //  that is an array of 2 items: [the error, domain]
    const failed = results.filter(isFail);
  }
)

Вы должны выяснить, что делает клиент redis, я пытался выяснить это с помощью документации, но могу также спросить мою золотую рыбку. После того, как вы перестроили поведение клиента, лучше всего попробовать с небольшим размером пакета, чтобы увидеть, есть ли какие-либо ошибки. Вы должны импортировать lib, чтобы использовать его, вы можете найти его здесь .

Я смог воспользоваться тем, что сказал Кевин Б, чтобы найти гораздо более быстрый способ запроса данных. Что я сделал, так это изменил запрос, чтобы я мог получить тренд для всех доменов от Афины. Я заказал его по domain_name, а затем отправил его в виде потока узлов, чтобы я мог отделить каждое доменное имя в его собственный JSON по мере поступления данных.

В любом случае, это то, с чем я закончил.

exports.storeDomainTrends = () => {
return new Promise((resolve, reject)=>{
    var streamObj = athenaClient.execute(`
    SELECT field,
            field,
            field,
            SUM(field) AS field
    FROM "db"."table"
    WHERE field IN ('Month')
    GROUP BY  field, field, field
    ORDER BY  field desc`).toStream();

    var data = [];

    streamObj.on('data', (record)=>{
        if (!data.length || record.field === data[0].field){
            data.push(record)
        } else if (data[0].field !== record.field){
            redisClient.set(('Key'), JSON.stringify(data))
            data = [record]
        }
    })

    streamObj.on('end', resolve);

    streamObj.on('error', reject);

})
.then()

}