Node.js 到 Flume-NG

Question

是否有一个功能性项目可以将数据从 Node.js 移动到 Flume-NG，而无需中间文件。

也许我遗漏了什么，我认为将数据从 Node.js 移动到 Flume 是一种更常见的需求，但似乎并非如此。

我发现许多项目似乎都在尝试这样做，但似乎都在大约 3 年前被放弃了，并且在当前版本中无法正常工作。似乎有一些可以与旧版本的flume一起使用，但是随着flume-ng的API发生了很大的变化，它们不再适用。

我找到了适用于 Node.js 的 avro 和 thrift 模块，并且 thrift 现在支持 node.js 似乎表明这应该是直截了当的，但这不起作用，可能没有足够的信息来说明要使用哪种传输/协议与 Flume-NG 一起使用，或者我只是不太了解它。

在我重新发明轮子之前，谁能指出我正确的方向？

这是我拥有的当前节点代码。它会生成一个 ECONNREFUSED。

#!/usr/local/bin/node

var thrift = require('thrift');
//var ThriftTransports = require('thrift/transport');
//var ThriftProtocols = require('thrift/protocol');

var Flume = require('./gen-nodejs/ThriftSourceProtocol');
var ttypes = require('./gen-nodejs/flume_types');

transport = thrift.TBufferedTransport();
protocol = thrift.TBinaryProtocol();
//transport = ThriftTransports.TBufferedTransport();
//protocol = ThriftProtocols.TBinaryProtocol();
var connection = thrift.createConnection("127.0.0.1", 51515,{
        transport: transport,
        protocol: protocol
});

connection.on('error', function(err) {
  console.error(err);
});

var client = thrift.createClient(Flume, connection);


var myEvent = new ttypes.ThriftFlumeEvent();
myEvent.headers = {};
myEvent.body = "body";

client.append(myEvent, function(err,data) {
  if (err) {
    // handle err
  } else {
    // data == [ttypes.ColumnOrSuperColumn, ...]
  }
  connection.end();
});

Answer 1

这是在FLUME-1894，文件ThriftSource.java中实现的Thrift Server的设置：

  args.protocolFactory(new TCompactProtocol.Factory());
  args.inputTransportFactory(new TFastFramedTransport.Factory());
  args.outputTransportFactory(new TFastFramedTransport.Factory());
  args.processor(new ThriftSourceProtocol.Processor<ThriftSourceHandler>(new ThriftSourceHandler()));

要让它工作，你需要在客户端使用兼容的堆栈：

• 紧凑协议
• 快速装帧运输

ThriftRpcClient.java已经有客户端实现了，不用再造轮子了。