hadoop - Flume:数据传输到服务器

Question

我是 Flume-ng 的新手。我必须编写一个程序，它可以将文本文件传输到其他程序(代理)。我知道我们必须了解代理，即主机 IP、端口号等。然后应该定义源、接收器和 channel 。我只想将日志文件传输到服务器。我的客户端代码如下。
公共(public)类 MyRpcClientFacade {

public class MyClient{

  private RpcClient client;
  private String hostname;
  private int port;

  public void init(String hostname, int port) {
        this.hostname = hostname;
        this.port = port;
        this.client = RpcClientFactory.getDefaultInstance(hostname, port);

      }

      public void sendDataToFlume(String data) {
        Event event = EventBuilder.withBody(data, Charset.forName("UTF-8"));
        try {
          client.append(event);
        } catch (EventDeliveryException e) {
          client.close();
          client = null;
          client = RpcClientFactory.getDefaultInstance(hostname, port);
        }
      }

      public void cleanUp() {
        client.close();
      }
}

以上代码只能发送 String数据到指定进程。但我必须发送文件。另外请告诉我是否 Source,Channel and Sink必须写入服务器吗？如果是这样，如何配置和编写这三个。请帮我。给出 Source,Sink And Channel的小样本

Answer 1

实际上，您只需要在每个节点上安装 Flume 客户端即可。然后，您提供一个配置文件，提供有关其行为的信息。
例如，如果您的节点读取文件(读取每个新行并将它们作为事件发送到 channel )，并通过 RPC 套接字发送文件内容。您的配置将如下所示:

  # sources/sinks/channels list
  <Agent>.sources = <Name Source1>
  <Agent>.sinks = <Name Sink1>
  <Agent>.channels = <Name Channel1> 
  # Channel attribution to a source
  <Agent>.sources.<Name Source1>.channels = <Name Channel1>
  # Channel attribution to sink
  <Agent>.sinks.<Name Sink1>.channels = <Name Channel1>
  # Configuration (sources,channels and sinks)
  # Source properties : <Name Source1>
  <Agent>.sources.<Name Source1>.type = exec
  <Agent>.sources.<Name Source1>.command = tail -F test
  <Agent>.sources.<Name Source1>.channels = <Name Channel1>
  # Channel properties : <Name Channel1>
  <Agent>.channels.<Name Channel1>.type = memory
  <Agent>.channels.<Name Channel1>.capacity = 1000
  <Agent>.channels.<Name Channel1>.transactionCapacity = 1000
  # Sink properties : <Name Sink1>
  <Agent>.sinks.<Nom Sink1>.type = avro
  <Agent>.sinks.<Nom Sink1>.channel = <Nom Channel1>
  <Agent>.sinks.<Nom Sink1>.hostname = <HOST NAME or IP>
  <Agent>.sinks.<Nom Sink1>.port = <PORT NUMBER>

然后你必须设置一个代理，它将读取同一端口上的 avro 源并以你想要存储它们的方式处理事件。
我希望它有帮助；)