Skip to content

fix(scrapers): Use RequestQueue directly to avoid excessive RQ writes on runs with large startUrls list #393

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
janbuchar merged 10 commits into from
May 21, 2025
Merged

fix(scrapers): Use RequestQueue directly to avoid excessive RQ writes on runs with large startUrls list #393

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
64e49ef
fix: Use RequestQueue directly to avoid excessive RQ writes on runs w…
janbuchar May 14, 2025
829fc5e
Remove thing
janbuchar May 14, 2025
2f8b8de
Update crawlee
janbuchar May 14, 2025
dcea0a3
Remove unused import
janbuchar May 15, 2025
40a809d
Format
janbuchar May 15, 2025
471fa88
Add flag to prevent repeated insertion of start requests
janbuchar May 20, 2025
9b792d0
Use the RQ-initialized flag in all scrapers
janbuchar May 21, 2025
c4f2ec5
Merge remote-tracking branch 'origin/master' into use-request-queue-d…
janbuchar May 21, 2025
938fd17
Resolve lint errors
janbuchar May 21, 2025
a759149
Revert package-lock changes
janbuchar May 21, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions package-lock.json
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,7 +77,7 @@
"@types/tough-cookie": "^4.0.5",
"@types/ws": "^8.5.12",
"commitlint": "^19.3.0",
"crawlee": "^3.13.0",
"crawlee": "^3.13.5",
"eslint": "^9.23.0",
"eslint-config-prettier": "^10.1.1",
"fs-extra": "^11.2.0",
Expand Down
43 changes: 33 additions & 10 deletions packages/actor-scraper/camoufox-scraper/src/internals/crawler_setup.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@ import type { Input } from './consts.js';
import { ProxyRotation } from './consts.js';

const SESSION_STORE_NAME = 'APIFY-PLAYWRIGHT-SCRAPER-SESSION-STORE';
const REQUEST_QUEUE_INIT_FLAG_KEY = 'REQUEST_QUEUE_INITIALIZED';

const { META_KEY, DEVTOOLS_TIMEOUT_SECS, SESSION_MAX_USAGE_COUNTS } =
scraperToolsConstants;
Expand Down Expand Up @@ -76,7 +77,6 @@ export class CrawlerSetup implements CrawlerSetupOptions {
requestQueueName?: string;

crawler!: PlaywrightCrawler;
requestList!: RequestList;
dataset!: Dataset;
pagesOutputted!: number;
private initPromise: Promise<void>;
Expand Down Expand Up @@ -167,7 +167,6 @@ export class CrawlerSetup implements CrawlerSetupOptions {

// Initialize async operations.
this.crawler = null!;
this.requestList = null!;
this.requestQueue = null!;
this.dataset = null!;
this.keyValueStore = null!;
Expand All @@ -182,21 +181,46 @@ export class CrawlerSetup implements CrawlerSetupOptions {
return req;
});

this.requestList = await RequestList.open(
'PLAYWRIGHT_SCRAPER',
startUrls,
);
// KeyValueStore
this.keyValueStore = await KeyValueStore.open(this.keyValueStoreName);

// RequestQueue
this.requestQueue = await RequestQueueV2.open(this.requestQueueName);

if (
!(await this.keyValueStore.recordExists(
REQUEST_QUEUE_INIT_FLAG_KEY,
))
) {
const requests: Request[] = [];
for await (const request of await RequestList.open(
null,
startUrls,
)) {
if (
this.input.maxResultsPerCrawl > 0 &&
requests.length >= 1.5 * this.input.maxResultsPerCrawl
) {
break;
}
requests.push(request);
}

const { waitForAllRequestsToBeAdded } =
await this.requestQueue.addRequestsBatched(requests);

void waitForAllRequestsToBeAdded.then(async () => {
await this.keyValueStore.setValue(
REQUEST_QUEUE_INIT_FLAG_KEY,
'1',
);
});
}

// Dataset
this.dataset = await Dataset.open(this.datasetName);
const info = await this.dataset.getInfo();
this.pagesOutputted = info?.itemCount ?? 0;

// KeyValueStore
this.keyValueStore = await KeyValueStore.open(this.keyValueStoreName);
}

/**
Expand All @@ -207,7 +231,6 @@ export class CrawlerSetup implements CrawlerSetupOptions {

const options: PlaywrightCrawlerOptions = {
requestHandler: this._requestHandler.bind(this),
requestList: this.requestList,
requestQueue: this.requestQueue,
requestHandlerTimeoutSecs: this.devtools
? DEVTOOLS_TIMEOUT_SECS
Expand Down
2 changes: 1 addition & 1 deletion packages/actor-scraper/cheerio-scraper/package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@
"type": "module",
"dependencies": {
"@apify/scraper-tools": "^1.1.4",
"@crawlee/cheerio": "^3.13.2",
"@crawlee/cheerio": "^3.13.4",
"apify": "^3.2.6"
},
"devDependencies": {
Expand Down
40 changes: 33 additions & 7 deletions packages/actor-scraper/cheerio-scraper/src/internals/crawler_setup.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@ const SCHEMA = JSON.parse(

const MAX_EVENT_LOOP_OVERLOADED_RATIO = 0.9;
const SESSION_STORE_NAME = 'APIFY-CHEERIO-SCRAPER-SESSION-STORE';
const REQUEST_QUEUE_INIT_FLAG_KEY = 'REQUEST_QUEUE_INITIALIZED';

/**
* Holds all the information necessary for constructing a crawler
Expand All @@ -70,7 +71,6 @@ export class CrawlerSetup implements CrawlerSetupOptions {
requestQueueName?: string;

crawler!: CheerioCrawler;
requestList!: RequestList;
dataset!: Dataset;
pagesOutputted!: number;
proxyConfiguration?: ProxyConfiguration;
Expand Down Expand Up @@ -151,7 +151,6 @@ export class CrawlerSetup implements CrawlerSetupOptions {

// Initialize async operations.
this.crawler = null!;
this.requestList = null!;
this.requestQueue = null!;
this.dataset = null!;
this.keyValueStore = null!;
Expand All @@ -167,19 +166,47 @@ export class CrawlerSetup implements CrawlerSetupOptions {
return req;
});

this.requestList = await RequestList.open('CHEERIO_SCRAPER', startUrls);
// KeyValueStore
this.keyValueStore = await KeyValueStore.open(this.keyValueStoreName);

// RequestQueue
this.requestQueue = await RequestQueueV2.open(this.requestQueueName);

if (
!(await this.keyValueStore.recordExists(
REQUEST_QUEUE_INIT_FLAG_KEY,
))
) {
const requests: Request[] = [];
for await (const request of await RequestList.open(
null,
startUrls,
)) {
if (
this.input.maxResultsPerCrawl > 0 &&
requests.length >= 1.5 * this.input.maxResultsPerCrawl
) {
break;
}
requests.push(request);
}

const { waitForAllRequestsToBeAdded } =
await this.requestQueue.addRequestsBatched(requests);

void waitForAllRequestsToBeAdded.then(async () => {
await this.keyValueStore.setValue(
REQUEST_QUEUE_INIT_FLAG_KEY,
'1',
);
});
}

// Dataset
this.dataset = await Dataset.open(this.datasetName);
const info = await this.dataset.getInfo();
this.pagesOutputted = info?.itemCount ?? 0;

// KeyValueStore
this.keyValueStore = await KeyValueStore.open(this.keyValueStoreName);

// Proxy configuration
this.proxyConfiguration = (await Actor.createProxyConfiguration(
this.input.proxyConfiguration,
Expand All @@ -197,7 +224,6 @@ export class CrawlerSetup implements CrawlerSetupOptions {
requestHandler: this._requestHandler.bind(this),
preNavigationHooks: [],
postNavigationHooks: [],
requestList: this.requestList,
requestQueue: this.requestQueue,
navigationTimeoutSecs: this.input.pageLoadTimeoutSecs,
requestHandlerTimeoutSecs: this.input.pageFunctionTimeoutSecs,
Expand Down
40 changes: 33 additions & 7 deletions packages/actor-scraper/jsdom-scraper/src/internals/crawler_setup.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@ const SCHEMA = JSON.parse(

const MAX_EVENT_LOOP_OVERLOADED_RATIO = 0.9;
const SESSION_STORE_NAME = 'APIFY-JSDOM-SCRAPER-SESSION-STORE';
const REQUEST_QUEUE_INIT_FLAG_KEY = 'REQUEST_QUEUE_INITIALIZED';

/**
* Holds all the information necessary for constructing a crawler
Expand All @@ -69,7 +70,6 @@ export class CrawlerSetup implements CrawlerSetupOptions {
requestQueueName?: string;

crawler!: JSDOMCrawler;
requestList!: RequestList;
dataset!: Dataset;
pagesOutputted!: number;
proxyConfiguration?: ProxyConfiguration;
Expand Down Expand Up @@ -150,7 +150,6 @@ export class CrawlerSetup implements CrawlerSetupOptions {

// Initialize async operations.
this.crawler = null!;
this.requestList = null!;
this.requestQueue = null!;
this.dataset = null!;
this.keyValueStore = null!;
Expand All @@ -166,19 +165,47 @@ export class CrawlerSetup implements CrawlerSetupOptions {
return req;
});

this.requestList = await RequestList.open('JSDOM_SCRAPER', startUrls);
// KeyValueStore
this.keyValueStore = await KeyValueStore.open(this.keyValueStoreName);

// RequestQueue
this.requestQueue = await RequestQueueV2.open(this.requestQueueName);

if (
!(await this.keyValueStore.recordExists(
REQUEST_QUEUE_INIT_FLAG_KEY,
))
) {
const requests: Request[] = [];
for await (const request of await RequestList.open(
null,
startUrls,
)) {
if (
this.input.maxResultsPerCrawl > 0 &&
requests.length >= 1.5 * this.input.maxResultsPerCrawl
) {
break;
}
requests.push(request);
}

const { waitForAllRequestsToBeAdded } =
await this.requestQueue.addRequestsBatched(requests);

void waitForAllRequestsToBeAdded.then(async () => {
await this.keyValueStore.setValue(
REQUEST_QUEUE_INIT_FLAG_KEY,
'1',
);
});
}

// Dataset
this.dataset = await Dataset.open(this.datasetName);
const info = await this.dataset.getInfo();
this.pagesOutputted = info?.itemCount ?? 0;

// KeyValueStore
this.keyValueStore = await KeyValueStore.open(this.keyValueStoreName);

// Proxy configuration
this.proxyConfiguration = (await Actor.createProxyConfiguration(
this.input.proxyConfiguration,
Expand All @@ -198,7 +225,6 @@ export class CrawlerSetup implements CrawlerSetupOptions {
runScripts: this.input.runScripts ?? true,
hideInternalConsole: !(this.input.showInternalConsole ?? false),
postNavigationHooks: [],
requestList: this.requestList,
requestQueue: this.requestQueue,
navigationTimeoutSecs: this.input.pageLoadTimeoutSecs,
requestHandlerTimeoutSecs: this.input.pageFunctionTimeoutSecs,
Expand Down
43 changes: 33 additions & 10 deletions packages/actor-scraper/playwright-scraper/src/internals/crawler_setup.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@ import type { Input } from './consts.js';
import { ProxyRotation } from './consts.js';

const SESSION_STORE_NAME = 'APIFY-PLAYWRIGHT-SCRAPER-SESSION-STORE';
const REQUEST_QUEUE_INIT_FLAG_KEY = 'REQUEST_QUEUE_INITIALIZED';

const {
META_KEY,
Expand Down Expand Up @@ -80,7 +81,6 @@ export class CrawlerSetup implements CrawlerSetupOptions {
requestQueueName?: string;

crawler!: PlaywrightCrawler;
requestList!: RequestList;
dataset!: Dataset;
pagesOutputted!: number;
private initPromise: Promise<void>;
Expand Down Expand Up @@ -198,7 +198,6 @@ export class CrawlerSetup implements CrawlerSetupOptions {

// Initialize async operations.
this.crawler = null!;
this.requestList = null!;
this.requestQueue = null!;
this.dataset = null!;
this.keyValueStore = null!;
Expand All @@ -213,21 +212,46 @@ export class CrawlerSetup implements CrawlerSetupOptions {
return req;
});

this.requestList = await RequestList.open(
'PLAYWRIGHT_SCRAPER',
startUrls,
);
// KeyValueStore
this.keyValueStore = await KeyValueStore.open(this.keyValueStoreName);

// RequestQueue
this.requestQueue = await RequestQueueV2.open(this.requestQueueName);

if (
!(await this.keyValueStore.recordExists(
REQUEST_QUEUE_INIT_FLAG_KEY,
))
) {
const requests: Request[] = [];
for await (const request of await RequestList.open(
null,
startUrls,
)) {
if (
this.input.maxResultsPerCrawl > 0 &&
requests.length >= 1.5 * this.input.maxResultsPerCrawl
) {
break;
}
requests.push(request);
}

const { waitForAllRequestsToBeAdded } =
await this.requestQueue.addRequestsBatched(requests);

void waitForAllRequestsToBeAdded.then(async () => {
await this.keyValueStore.setValue(
REQUEST_QUEUE_INIT_FLAG_KEY,
'1',
);
});
}

// Dataset
this.dataset = await Dataset.open(this.datasetName);
const info = await this.dataset.getInfo();
this.pagesOutputted = info?.itemCount ?? 0;

// KeyValueStore
this.keyValueStore = await KeyValueStore.open(this.keyValueStoreName);
}

/**
Expand All @@ -241,7 +265,6 @@ export class CrawlerSetup implements CrawlerSetupOptions {

const options: PlaywrightCrawlerOptions = {
requestHandler: this._requestHandler.bind(this),
requestList: this.requestList,
requestQueue: this.requestQueue,
requestHandlerTimeoutSecs: this.devtools
? DEVTOOLS_TIMEOUT_SECS
Expand Down
Loading
Loading