flume学习05---Thrift Source

Thrift Source

侦听Thrift端口并从外部Thrift客户端流接收事件。 当与另一（前一跳）Flume agent上的内置ThriftSink配对时，它可以创建分层集合拓扑。 Thrift源可以配置为通过启用kerberos身份验证在安全模式下启动。 agent-principal和agent-keytab是Thrift源用来向Kerberos KDC进行身份验证的属性。 必需属性以粗体显示。 Example for agent named a1:

a1.sources = r1
a1.channels = c1
a1.sources.r1.type = thrift
a1.sources.r1.channels = c1
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 4141

二、实现 2.1、配置文件， source type为thrift,监听主机master的4141端口。

[root@master testconf]# cat thrift.properties 
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
a1.sources.r1.type=thrift
a1.sources.r1.bind=master
a1.sources.r1.port=4141

# Describe the sink
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink  
a1.sinks.k1.topic = testflume  
a1.sinks.k1.brokerList = master:9092,slave1:9092,slave2:9092  
a1.sinks.k1.requiredAcks = 1  
a1.sinks.k1.batchSize = 20  

# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000000
a1.channels.c1.transactionCapacity = 10000

# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

[root@master testconf]#

2.2、通过thrift客户断向flume写数据 获取客户端`this.client = RpcClientFactory.getThriftInstance(hostname, port);

`

package com.chb.flume;

import java.nio.charset.Charset;

import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.api.RpcClient;
import org.apache.flume.api.RpcClientFactory;
import org.apache.flume.event.EventBuilder;

class MyRpcClientFacade {
    private RpcClient client;
    private String hostname;
    private int port;

    public void init(String hostname, int port) {
        // Setup the RPC connection
        this.hostname = hostname;
        this.port = port;
        //this.client = RpcClientFactory.getDefaultInstance(hostname, port);
        //Use the following method to create a thrift client (instead of the above line):
        this.client = RpcClientFactory.getThriftInstance(hostname, port);

    }

    public void sendDataToFlume(String data) {
        // Create a Flume Event object that encapsulates the sample data
        Event event = EventBuilder.withBody(data, Charset.forName("UTF-8"));
        // Send the event
        try {
            client.append(event);
        } catch (EventDeliveryException e) {
            // clean up and recreate the client
            client.close();
            client = null;
            //client = RpcClientFactory.getDefaultInstance(hostname, port);
            //Use the following method to create a thrift client (instead of the above line):
            this.client = RpcClientFactory.getThriftInstance(hostname, port);
        }
    }

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

具体执行， 向master的4141端口写数据

package com.chb.flume;

public class MyTest {
    public static void main(String[] args) {
        MyRpcClientFacade client = new MyRpcClientFacade();
        //初始化client
        client.init("192.168.10.224", 4141);
        String sampleData = "Hello Flume!";
        System.out.println(sampleData);
        for (int i = 0; i < 10; i++) {
            long a = System.currentTimeMillis();
            System.out.println(a);
            //向flume写数据
            client.sendDataToFlume(sampleData);
        }
        client.cleanUp();
    }
}

kafka消费