All files / Nodejs/test fast-s3-read-test.js

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
// USAGE
// node fast-s3-read-test test|prod fast|slow
 
let env = {
	test: {
		LeoStream: "TestBus-LeoStream-R2VV0EJ6FRI9",
		LeoCron: "TestBus-LeoCron-OJ8ZNCEBL8GM",
		LeoSettings: "TestBus-LeoSettings-YHQHOKWR337E",
		LeoEvent: "TestBus-LeoEvent-FNSO733D68CR",
		LeoSystem: "TestBus-LeoSystem-L9OY6AV8E954",
		LeoArchive: "TestBus-LeoArchive-WUWG7N8OXG97",
		LeoKinesisStream: "TestBus-LeoKinesisStream-1XY97YYPDLVQS",
		LeoFirehoseStream: "TestBus-LeoFirehoseStream-1M8BJL0I5HQ34",
		LeoS3: "testbus-leos3-1erchsf3l53le",
		Region: "us-east-1"
	},
	prod: {
		"LeoArchive": "ProdBus-LeoArchive-K9UQKPSB7M93",
		"LeoCron": "ProdBus-LeoCron-CNT18F32S1UK",
		"LeoEvent": "ProdBus-LeoEvent-12AQ6PZNRRHNK",
		"LeoFirehoseStream": "ProdBus-LeoFirehoseStream-1GZRLNQ9YN9BK",
		"LeoKinesisStream": "ProdBus-LeoKinesisStream-1SZPWFHEKF669",
		"LeoS3": "prodbus-leos3-17uqyaemyrrxs",
		"LeoSettings": "ProdBus-LeoSettings-6FLTMJUIOS7Q",
		"LeoStream": "ProdBus-LeoStream-1RWIW0AV7AC0Y",
		"LeoSystem": "ProdBus-LeoSystem-1J1ZWZIGL7M3F",
		"Region": "us-east-1"
	}
};
var leo = require("../index.js")(env[process.argv[2]]);
console.log("ENV:", process.argv[2]);
let ls = leo.streams;
let start = Date.now();
let last = start;
let records = 0;
 
let is_fast = process.argv[3] == "fast";
 
let eid;
ls.pipe(
	leo.read("JUNK", "item-entity-old-new", {
		start: "z/2021/11/18/23/00/",
		//limit: 50000,
		fast_s3_read: is_fast,
		fast_s3_read_parallel_fetch_max_bytes: 1024 * 1024 * 100,
		stream_query_limit: 10000,
		//loops: Number.POSITIVE_INFINITY
		stopTime: Date.now() + 60000
	}),
	ls.through((event, done) => {
		eid = event.eid;
		//console.log(JSON.stringify(event));
		records++;
		if (records % 1000 == 0) {
			let duration = Date.now() - start;
			let otherduration = Date.now() - last;
			last = Date.now();
			console.log(`Records: ${records}, Duration: ${duration}ms, RPS: ${Math.floor(records / (duration / 1000))}, 1000 => Duration: ${otherduration}ms, RPS: ${Math.floor(1000 / (otherduration / 1000))}`);
		}
		done();
	}),
	ls.devnull(),
	(error) => {
		let duration = Date.now() - start;
		console.log("Done:", is_fast ? "fast" : "slow", eid, error || "");
		console.log(`Records: ${records}, Duration: ${duration}ms, RPS: ${records / (duration / 1000)}`);
	}
);
 
 
// Slow with 50 fetch
//Done: slow z/2021/09/09/15/37/1631201860845-0000002
//Records: 9390, Duration: 240103ms, RPS: 39.10821605727542
 
// Fast with 50 fetch
//Done: fast z/2021/09/24/22/37/1632523060584-0000002
//Records: 12602, Duration: 42810ms, RPS: 294.37047418827376
 
// Fast with 400 fetch
//Done: fast z/2021/09/24/22/37/1632523060584-0000002
//Records: 12602, Duration: 14064ms, RPS: 896.0466439135381