Сколько запросов может Node-Express запустить одновременно?
У меня есть скрипт, который извлекает 25 000 записей из AWS Athena, которая в основном является реляционной базой данных SQL PrestoDB. Предположим, что я генерирую запрос для каждой из этих записей, что означает, что я должен сделать 25 000 запросов к Athena, а затем, когда данные возвращаются, я должен сделать 25 000 запросов к моему кластеру Redis.
Каким было бы идеальное количество запросов, которые можно сделать за один раз от узла к Афине?
Причина, по которой я спрашиваю, заключается в том, что я пытался сделать это, создавая массив из 25 000 обещаний, а затем вызов Promise.all(promiseArray)
на нем, но приложение просто зависло навсегда.
Проблема в том, что это занимает целую вечность. Я взял примерно часовой перерыв и вернулся, и там осталось 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 ответа:
Используя 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()
}