Flume简介与使用（二）——Thrift Source采集数据

2023-07-27 22:56:28

　　继上一篇安装Flume后，本篇将介绍如何使用Thrift Source采集数据。

　　Thrift是Google开发的用于跨语言RPC通信，它拥有功能强大的软件堆栈和代码生成引擎，允许定义一个简单的IDL文件来生成不同语言的代码，服务器端和客户端通过共享这个IDL文件来构建来完成通信。

　　Flume的Thrift Source是其实现的众多Source中的一个，Flume已经实现了服务器端，因此我们可以用任意自己熟悉的语言编写自己的Thrift Source客户端来采集数据，然后发送给Thrift Source服务器端。

　　[一]、生成C++代码

　　下载源码版的Flume，在apache-flume-1.6.0-src\flume-ng-sdk\src\main\thrift目录下有Flume定义好的flume.thrift文件，现在只要用这个文件来生成我们需要的代码就行了。

　　flume.thrift文件内容如下：

 namespace java org.apache.flume.thrift

 struct ThriftFlumeEvent {

   1: required map <string, string> headers,

   2: required binary body,

 }

 enum Status {

   OK,

   FAILED,

   ERROR,

   UNKNOWN

 }

 service ThriftSourceProtocol {

   Status append(1: ThriftFlumeEvent event),

   Status appendBatch(1: list<ThriftFlumeEvent> events),

 }

　　1、定义了一个ThriftFlumeEvent结构体，用来封装发送的数据；

　　2、定义了一个service类ThriftSourceProtocol，服务器端具体实现ThriftSourceProtocol里面的两个方法，再由客户端调用这些方法把数据传给Thrift Source服务器端。

　　3、运行下面的命令：thrift --gen cpp flume.thrift，会在当前目录生成gen-cpp目录，里面是Thrift自动生成c++头文件和代码。（在这之前要先安装Thrift）

　　[二]、下面是编写自己的客户端代码，我这里是接收远程传过来的数据，然后发送给Flume的Thrift Source服务器。

 #include <arpa/inet.h>

 #include <sys/types.h>

 #include <sys/socket.h>

 #include <pthread.h>

 #include <unistd.h>

 #include <stdlib.h>

 #include "include/MESA_prof_load.h"

 #include "include/MESA_handle_logger.h"

 #include <string>

 #include <iostream>

 #include "gen-cpp/flume_constants.h"

 #include "gen-cpp/flume_types.h"

 #include "gen-cpp/ThriftSourceProtocol.h"

 #include <thrift/protocol/TBinaryProtocol.h>

 #include <thrift/protocol/TCompactProtocol.h>

 #include <thrift/transport/TSocket.h>

 #include <thrift/transport/TTransportUtils.h>

 using namespace std;

 using namespace apache::thrift;

 using namespace apache::thrift::protocol;

 using namespace apache::thrift::transport;

 #define LOG_PATH "/home/zjf/DFcode/trafficlog/traffic_source.log"

 #define DATA_BUFFER 2048    //send buffer data length

 #define BUFLEN   2048       //received buffer data length

 #define BATCH_SIZE 1000     //send event num to flume once

 //defined my C++ object

 class ThriftClient{

     public:

         // Thrift protocol needings...

         boost::shared_ptr<TTransport> socket;

         boost::shared_ptr<TTransport> transport;

         boost::shared_ptr<TProtocol> protocol;

         ThriftSourceProtocolClient* pClient;

     public:

         ThriftClient();

 };

 //cconstruction function, init the thrift source server ip and port

 ThriftClient::ThriftClient():

     socket(new TSocket("10.208.129.12",)),

     transport(new TFramedTransport(socket)),

     protocol(new TCompactProtocol(transport))

 {

     pClient = new ThriftSourceProtocolClient(protocol);

 }

 //log

 struct log_info_t{

     char *path;

     int log_level;

     void * handle;

 };

 struct log_info_t log_info;

 const char *module = "zjf_traffic_data_collector";

 //类的对象

 ThriftClient *client = new ThriftClient();

 std::map<std::string, std::string>  headers;

 std::vector<ThriftFlumeEvent> eventbatch;

 unsigned long long pkt_num_tgl = ;

 int RecvAndSendUDP(){

     MESA_handle_runtime_log(log_info.handle, RLOG_LV_INFO, module, "RecvUDP be called");

     int listen_socket;　　　　　　　　　　//socket id

     struct sockaddr_in    local;　　　　//client IP, where to recevied data

     struct sockaddr_in    from;    　　//server IP(local host)

     char server_addr[] = "10.208.129.12";    //received traffic IP

     int server_port = ;                    //received traffic port

     char send_buf[DATA_BUFFER] = {};          //data send to flume

     char Buf[BUFLEN] = {};

     int fromlen;

     int len;

     //init socket

 reconnect:

     memset(&local, , sizeof(local));

     local.sin_family = AF_INET;

     local.sin_addr.s_addr = inet_addr(server_addr);

     local.sin_port = htons(server_port);

     listen_socket = socket(AF_INET, SOCK_DGRAM, ); // UDP socket

     if(listen_socket < ) {

         printf("error udp socket\n");

     }else{

         printf("listen_socket create OK\n");

     }

     if(bind(listen_socket, (struct sockaddr *)&local, sizeof(local)) < ) {

         printf("error udp bind\n");

         return -;

     }else{

         printf("socket bind OK\n");

     }

     while(){

         char sip[] = {};

         char dip[] = {};

         char srcport[] = {};

         char destport[] = {};

         char url[BUFLEN] = {};

         memset(Buf,,BUFLEN);

         fromlen = sizeof(from);

         len = recvfrom(listen_socket, (void *)Buf, (size_t)BUFLEN, , (struct sockaddr *)&from,(socklen_t *)&fromlen);

         if(len == -) {

             printf("error udp recvfrom\n");

             close(listen_socket);

             goto reconnect;

         }

         //parse received buf, transform to key-value

         int i;

         int sip_loc = ;

         int sport_loc = ;

         int dip_loc = ;

         int dport_loc = ;

         int dotcount = ;

         for(i=;Buf[i] != '\0';i++){

             if(Buf[i] == '.'){

                 dotcount++;

                 if(dotcount == ){

                     sip_loc = i;

                     memcpy(sip,Buf,i);

                 }

                 else if(dotcount == ){

                     dip_loc = i;

                     memcpy(dip,Buf+sport_loc+,dip_loc-sport_loc-);

                 }

                 else if(dotcount == ){

                     dport_loc = i;

                     memcpy(destport,Buf+dip_loc+,dport_loc-dip_loc-);

                     break;

                 }

                 else{}

             }

             if(Buf[i] == '>'){

                 sport_loc = i;

                 memcpy(srcport,Buf+sip_loc+,sport_loc-sip_loc-);

             }

         }

         memcpy(url,Buf+dport_loc+,strlen(Buf)-dport_loc);

         unsigned long src_ip = inet_addr(sip);

         unsigned long dst_ip = inet_addr(dip);

         sprintf(send_buf,"SrcIP=%u SrcPort=%s DestIP=%u DestPort=%s",ntohl(src_ip),srcport,ntohl(dst_ip),destport);

         //construct an event and append to send

         if( != strlen(send_buf) ){

             pkt_num_tgl++;

             string sBody(send_buf);

             ThriftFlumeEvent tfEvent;

             tfEvent.__set_headers(headers);

             tfEvent.__set_body(sBody);

             eventbatch.push_back(tfEvent);

             if(eventbatch.size() >= BATCH_SIZE){

                 if(!client->transport->isOpen())

                     client->transport->open();

                 Status::type res = client->pClient->appendBatch(eventbatch);

                 if(res != Status::OK){

                     MESA_handle_runtime_log(log_info.handle, RLOG_LV_FATAL, module, "WARNING: send event via thrift failed, return code:%d",res);

                 }else{

                     //printf("sended %lld event data to flume successful\n", pkt_num_tgl);

                 }

                 eventbatch.clear();

             }

         }

         bzero(send_buf,DATA_BUFFER);

     }

 }

 int main()

 {

     //create――logger

     log_info.path = (char *)LOG_PATH;

     log_info.log_level = ;

     log_info.handle = MESA_create_runtime_log_handle(log_info.path, log_info.log_level);

     //open thrift connection

     if(!client->transport->isOpen()){

         client->transport->open();

     }

     eventbatch.clear();

     RecvAndSendUDP();

     return ;

 }

[三]、编译并运行

　　g++ -g -DHAVE_NETINET_IN_H -I. -I/usr/local/include/thrift -L/usr/local/lib rec_send_traffic_thrift.cpp gen-cpp/flume_constants.cpp gen-cpp/flume_types.cpp gen-cpp/ThriftSourceProtocol.cpp -o rec_send_traffic_thrift -lthrift -lpcap -L/usr/lib64 -lMESA_htable -lpthread -lMESA_handle_logger

　　用守护进程启动程序：

 #!/bin/sh

 while [  ]; do

     ulimit -c unlimited

     #./jz

     #cgexec -g cpu,memory:/MESA/jz ./jz >> jz.log

     ./rec_send_traffic_thrift

     #./jz

     echo program crashed, restart at `date +"%w %Y/%m/%d, %H:%M:%S"` >> RESTART.log

     sleep

 done

推荐博文：【1】http://www.micmiu.com/soa/rpc/thrift-sample/

　　　　　【2】http://www.mamicode.com/info-detail-869223.html

　　　　　【3】http://blog.csdn.net/yuzx2008/article/details/50179033

　　　　　【4】http://shiyanjun.cn/archives/456.html

　　　　　【5】http://flume.apache.org/FlumeDeveloperGuide.html#rpc-clients-avro-and-thrift

转载请注明原文出处，谢谢

码农公寓

相关文章