node.js中一个大的json日志文件分析

Question

我有以下 JSON 文件：

sensorlogs.json
{"arr":[{"UTCTime":10000001,"s1":22,"s2":32,"s3":42,"s4":12},
{"UTCTime":10000002,"s1":23,"s2":33,"s4":13},
{"UTCTime":10000003,"s1":24,"s2":34,"s3":43,"s4":14},
{"UTCTime":10000005,"s1":26,"s2":36,"s3":44,"s4":16},
{"UTCTime":10000006,"s1":27,"s2":37,"s4":17},
{"UTCTime":10000004,"s1":25,"s2":35,"s4":15},
...
{"UTCTime":12345678,"s1":57,"s2":35,"s3":77,"s4":99}
]}

传感器 s1、s2、s3 等都以不同的频率传输（请注意，s3 每 2 秒传输一次，并且时间戳可能是无序的）。

我怎样才能实现类似 -

Analyzing s1:
s = [[10000001, 22], [10000002, 23],.. [12345678,57]]
s1 had 2 missing entries
Analyzing s2:
s = [[10000001, 32], [10000002, 33],.. [12345678,35]]
s2 had 0 missing entries
Analyzing s3:
s = [[10000001, 42], [10000003, 43],.. [12345678,77]]
s3 had 0 missing entries
Analyzing s4:
s = [[10000001, 12], [10000003, 13],.. [12345678,99]]
s4 had 1 missing entries

sensorlogs.json 为 16 GB。

可以根据连续 UTC 时间戳的差异找到丢失的条目。每个传感器都以已知频率传输。

由于内存限制，我无法使用多个大型数组进行分析，因此我必须对同一个 JSON 日志文件进行多次传递，并且仅使用单个大型数组进行分析。

我现在所拥有的是关注 -

var result = [];
//1. Extract all the keys from the log file
console.log("Extracting keys... \n");
var stream = fs.createReadStream(filePath);
var lineReader = lr.createInterface(
{
  input: stream
});

lineReader.on('line', function (line) 
{
  getKeys(line);//extract all the keys from the JSON
});
stream.on('end', function()
{
  //obj -> arr
  for(var key in tmpObj)
    arrStrm.push(key);

  //2. Validate individual sensors
  console.log("Validating the sensor data ...\n");

  //Synchronous execution of the sensors in the array
  async.each(arrStrm, function(key)
  {
    {
        currSensor = key;
        console.log("validating " + currSensor + "...\n");

        stream = fs.createReadStream(filePath);
        lineReader = lr.createInterface(
        {
          input: stream
        });

        lineReader.on('line', function (line) 
        {
          processLine(line);//Create the arrays for the sensors
        });
        stream.on('end', function()
        {
            processSensor(currSensor);//Process the data for the current sensor
        });
    }
  });
});

function getKeys(line) 
{
    if(((pos = line.indexOf('[')) >= 0)||((pos = line.indexOf(']')) >= 0))
        return;
    if (line[line.length-1] == '\r') line=line.substr(0,line.length-1); // discard CR (0x0D)
    if (line[line.length-1] == ',') line=line.substr(0,line.length-1); // discard ,
//  console.log(line);

    if (line.length > 1) 
    { // ignore empty lines
        var obj = JSON.parse(line); // parse the JSON
        for(var key in obj) 
        {
            if(key != "debug")
            {
                if(tmpObj[key] == undefined)
                    tmpObj[key]=[];
            }
        };
    }
}

当然这不起作用，我在网上找不到任何解释如何实现的东西。

注意： 我可以选择我选择的任何语言来开发这个工具（C/C++、C#/Java/Python），但我选择使用 JavaScript，因为它能够轻松解析 JSON 数组（而且我也有兴趣在 JS 中变得更好） ）。 如果 JavaScript 不是最好的语言来制作这样的工具，有人愿意建议一种替代语言吗？

编辑：一些不是很清楚或者我之前没有包含的重要信息，但看起来在问题中包含很重要 -

1. JSON 日志中的数据不是实时流式传输，而是存储在硬盘中的 JSON 文件
2. 存储的数据不是按时间顺序排列的，这意味着时间戳的顺序可能不正确。因此，每个传感器数据都需要根据时间戳在存储在数组中
进行排序
3. 我不能为每个传感器使用单独的数组（这与在 RAM 中存储整个 16 GB JSON 相同），为了节省内存，一次只能使用一个数组。是的，我的日志中有超过 4 个传感器，这只是一个示例（大概有 20 个）

我已经修改了我的 JSON 和预期的输出

一种解决方案可能是对 JSON 文件进行多次传递，一次将一个带有时间戳的传感器数据存储在一个数组中，然后对数组进行排序，最后分析数据是否存在损坏和间隙。这就是我在上面的代码中尝试做的事情

Answer 1

所以你有 16GB 的大脂肪传感器日志包装在 json 中。

首先，一个 16GB 的整个 json 文件是不现实的，因为左括号和右括号打破了规律，变成了数组中的烦人字符。我们知道文件有开头和结尾，此外，没有它们，您的程序可以处理文件的块，甚至可以处理直接插入设备的流。所以让我们假设我们将要处理的是这样的：

{"UTCTime":10000001,"s1":22,"s2":32,"s3":42,"s4":12},
{"UTCTime":10000002,"s1":23,"s2":33,"s4":13},
{"UTCTime":10000003,"s1":24,"s2":34,"s3":43,"s4":14},
...
{"UTCTime":12345678,"s1":57,"s2":35,"s3":77,"s4":99},

甚至在末尾添加或检测缺少的逗号应该不会太难。

现在每一行都以相同的方式格式化，并且可以解释为 json。 问题是：传感器是否按预期输出数据？如果我们确定他们在正确的时间和正确的频率说话（案例 1）但有时他们可能会错过写作，那么一切都很好。但是，如果它们在时间范围内开始出现轻微滑移（案例 2），则需要某种启发式方法来恢复正确的线线频率，并且分析时间会更长。

如果我们不实时处理这个，首先对文件进行简单的验证检查是判断是否在每个freq 行上找到了预期的传感器数据，对吗？

无论如何，由于它是一个非常大的文件，因此必须尽可能逐行处理。

在下面的程序中，我只考虑了第一种情况，我们可以处理一个连续的流。

#!/usr/bin/python
import json

sensors={}
sensors['s1']=[1] # frequencies
sensors['s2']=[1]
sensors['s3']=[2]
sensors['s4']=[1]

# append data array and error counter at sensors[i]
# it holds [freq,err,data]
for k,v in sensors.iteritems(): sensors[k].extend([0,[]])
FRQ=0;ERR=1;DAT=2
print list(sorted(sensors.items()))
S=list(sorted(sensors.keys()))

with open('./sensors.json', "r") as stream:
    i=0
    for line in stream:
      if not line.rstrip(): continue # skip blank lines
      j=json.loads(line[:-2]) # skip comma and \n
      t=j["UTCTime"]
      for k in S:
          sensor=sensors[k]
          if i%sensor[FRQ]==0 : # every Nth iteration
            v=j.get(k)
            if v is None:
                sensor[ERR]+=1
                print k,"has",sensor[ERR],"missing entries"
            sensor[DAT].append([t,v]) # append that sensor data
            # filling up the memory...
      i+=1

for k,v in sorted(sensors.iteritems()): print k,sensors[k][DAT]
for k,v in sorted(sensors.iteritems()): print k,'had',sensors[k][ERR],"missing entries"

为了处理案例 2，我们将使用模数检查反转 None 检查，验证传感器是否在不应该写入的情况下写入了某些内容，然后尝试检测移位。

最后一点：您的程序可能会出现内存不足，因此将整个数据保存在内存中可能不是一个好主意。如果打算为每个传感器使用单独的数组进行进一步处理，则将它们写入文件可能更明智。

Answer 2

再次编辑以考虑您的编辑：

var fs = require('fs');
var stream = fs.createReadStream('sensorlogs.json', {flags: 'r', encoding: 'utf-8'});
var buffer = '';
var sensor = process.argv[2];
var readings = [];
var missingCont = 0;
console.log('Analizying ' + sensor + ':');

stream.on('data', function(d) {
    buffer += d.toString();
    processBuffer();
    console.log(readings);
    console.log(sensor + ' had ' + missingCont + ' missing entries');
});

function processBuffer() {
  buffer = buffer.slice(buffer.indexOf('[{'));
  while(buffer.indexOf('{') != -1) {
    buffer = buffer.slice(buffer.indexOf('{"'));
    processLine(buffer.slice(0, buffer.indexOf('}') + 1));
    buffer = buffer.slice(buffer.indexOf('}') + 2);
  }
};

function processLine(line) {
  if(line != ""){
    var obj = JSON.parse(line);
    if(!obj[sensor]){
      missingCont++;
    }else{
      var pos;
      for(pos = 0; pos < readings.length; pos++){
        if(obj.UTCTime < readings[pos][0]){
          var reading = [obj.UTCTime, obj[sensor]]
          readings.splice(pos, 0, reading);
          break;
        }
      }
      if(pos == readings.length){
        readings.push([obj.UTCTime, obj[sensor]]);
      }
    }
  }
};

您必须使用要分析的传感器的参数来调用它：

node.exe scripts\processJson.js <param>

为了测试它，我拿了这个样本：

{"arr":[{"UTCTime":10000001,"s1":22,"s2":32,"s3":42,"s4":12},
{"UTCTime":10000005,"s1":20,"s2":30,"s3":40,"s4":10},
{"UTCTime":10000002,"s1":23,"s2":33,"s4":13},
{"UTCTime":10000003,"s1":24,"s2":34,"s3":43,"s4":14},
{"UTCTime":12345678,"s1":57,"s2":35,"s3":77,"s4":99}
]}

输出是：

> node.exe scripts\processJson.js s1
Analizying s1:
[[10000001, 22], [10000002, 23], [10000003, 24], [10000005, 20], [12345678, 57]]
s1 had 0 missing entries

> node.exe scripts\processJson.js s2
Analizying s2:
[[10000001, 32], [10000002, 33], [10000003, 34], [10000005, 30], [12345678, 35]]
s2 had 0 missing entries

> node.exe scripts\processJson.js s3
Analizying s3:
[[10000001, 42], [10000003, 43], [10000005, 40], [12345678, 77]]
s3 had 1 missing entries

> node.exe scripts\processJson.js s4
Analizying s4:
[[10000001, 12], [10000002, 13], [10000003, 14], [10000005, 10], [12345678, 99]]
s4 had 0 missing entries