1

是否有一个功能项目可以将数据从 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();
});
4

1 回答 1

1

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

  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 中已经有客户端实现，因此您无需再次发明轮子。

于 2014-04-30T08:44:13.193 回答