-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathstream_txt_to_db_one_pause.js
67 lines (55 loc) · 2.02 KB
/
stream_txt_to_db_one_pause.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
var fs = require('fs'), es = require("event-stream"), MongoClient = require("mongodb").MongoClient;
var url = "mongodb://127.0.0.1:27017/";
MongoClient.connect(url, { useUnifiedTopology: true }, function (err, db) {
if (err) throw err;
console.time("dbsave");
var connection = db.db('netflix');
var movieId;
var s = fs.createReadStream('Files/combined_data_1.txt')
.pipe(es.split())
.pipe(es.mapSync(function (line) {
s.pause();
if (line.includes(':')) {
movieId = (line.split(':'))[0];
s.resume();
return;
}
line = line.split(',');
connection.collection('ratings').insertOne({
movieId: parseInt(movieId),
UserId: parseInt(line[0]),
rating: parseInt(line[1]),
date: new Date(line[2])
});
s.resume();
}).on('error', function (err) {
console.log('Error while reading file.', err);
db.close();
}).on('end', function () {
console.log('File read completed');
db.close();
console.timeEnd("dbsave");
}));
});
//----------------------------------------------------------------------------------------------------------------------
// var myInterface = readline.createInterface({
// input: fs.createReadStream('Files/combined_data_1.txt')
// });
// var lineno = 0;
// var id;
// //var obj = [];
// myInterface.on('line', function (line) {
// lineno++;
// if (line.includes(':')) {
// id = (line.split(':'))[0];
// return;
// }
// line = line.split(',');
// var obj = { movieIdentifier: id, userIndentifier: line[0], rating: line[1], date: line[2] };
// fs.appendFile('Files/combined_data_1.json', JSON.stringify(obj), function (err) {
// if (err) throw err;
// });
// console.log('Line number ' + lineno + ': ' + line);
// }).on('close', function (line) {
// console.log('done');
// });