Hadoop生态架构之LogServer实现

深藏阁楼爱情的钟 2023-06-29 08:59 22阅读 0赞

分布式日志数据采集

##### 1.NGINX配置： #####

slave2作为分发节点，master和slave1作为日志采集节点  
slave2的NGINX配置如下图：  
![format_png][]  
slave1和master的NGINX配置如下图：  
![format_png 1][]

##### 2.slave1和master的flume作为客户端的配置： #####

flume client端配置文件:flume-client.properties：

# 定义这个agent中各组件的名字
    a1.sources = r1
    a1.sinks = k1
    a1.channels = c1
    
    # set channel
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    
    # 描述和配置source组件：r1
    a1.sources.r1.type = exec
    a1.sources.r1.command = tail -F /usr/local/logserver_study/cgi_demo/logs/client.ERROR
    
    # 描述和配置sink组件：k1
    a1.sinks.k1.type = avro
    a1.sinks.k1.hostname = master
    a1.sinks.k1.port = 52020
    
    # 描述和配置source  channel   sink之间的连接关系
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

##### 3.master作为flume服务端的配置： #####

flume server端配置文件:flume-server.properties:

# 定义这个agent中各组件的名字
    a1.channels = c1
    a1.sources = r1
    a1.sinks = k1
    
    # set channel
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    
    a1.sources.r1.type = avro
    a1.sources.r1.bind = master
    a1.sources.r1.port = 52020
    
    a1.sinks.k1.type=hbase
    a1.sinks.k1.table=user_action_table
    a1.sinks.k1.columnFamily=action_log
    a1.sinks.k1.serializer=com.badou.FlumeHbaseEventSerializer
    a1.sinks.k1.serializer.columns=userid,itemid,type,ip
    
    
    # 描述和配置source  channel   sink之间的连接关系
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

##### 4.Java代码目录如下图： #####

![format_png 2][]

\]\# ./bin/flume-ng agent --conf conf --conf-file ./conf/logserver\_flume\_hbase/flume-client.properties --name a1 -Dflume.root.logger=INFO,console  
master:  
\]\# ./bin/flume-ng agent --conf conf --conf-file ./conf/logserver\_flume\_hbase/flume-client.properties --name a1 -Dflume.root.logger=INFO,console  
master:  
\]\# ./bin/flume-ng agent --conf conf --conf-file ./conf/logserver\_flume\_hbase/flume-server.properties --name a1 -Dflume.root.logger=INFO,console  
master:  
\]\# hbase shell  
\]\# create ‘user\_action\_table’,‘action\_log’

##### 5.Java具体代码及配置如下 #####

1.pom配置

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0"
             xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
    
        <groupId>org.example</groupId>
        <artifactId>FlumehHase</artifactId>
        <version>1.0-SNAPSHOT</version>
    
        <dependencies>
            <dependency>
                <groupId>org.apache.flume</groupId>
                <artifactId>flume-ng-core</artifactId>
                <version>1.6.0</version>
            </dependency>
    
            <dependency>
                <groupId>org.apache.flume.flume-ng-sinks</groupId>
                <artifactId>flume-ng-hbase-sink</artifactId>
                <version>1.6.0</version>
            </dependency>
    
            <dependency>
                <groupId>org.apache.hbase</groupId>
                <artifactId>hbase-client</artifactId>
                <version>0.98.6.1-hadoop2</version>
            </dependency>
            
            
    
    
        </dependencies>
    
        <build>
            <sourceDirectory>src/main/java</sourceDirectory>
            <testSourceDirectory>src/test/java</testSourceDirectory>
            <plugins>
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-compiler-plugin</artifactId>
                    <configuration>
                        <source>6</source>
                        <target>6</target>
                    </configuration>
                </plugin>
    
                <plugin>
                    
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-compiler-plugin</artifactId>
                    <version>3.3</version>
                    <configuration>
                        <source>1.7</source>
                        <target>1.7</target>
                        <encoding>UTF-8</encoding>
                    </configuration>
                </plugin>
    
                <plugin>
                    <groupId>org.scala-tools</groupId>
                    <artifactId>maven-scala-plugin</artifactId>
                    <version>2.15.2</version>
                    <executions>
                        <execution>
                            <goals>
                                <goal>compile</goal>
                                <goal>testCompile</goal>
                            </goals>
                        </execution>
                    </executions>
                </plugin>
    
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-assembly-plugin</artifactId>
                    <version>3.0.0</version>
                    <configuration>
                        <descriptorRefs>
                            <descriptorRef>jar-with-dependencies</descriptorRef>
                        </descriptorRefs>
                    </configuration>
                </plugin>
    
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-surefire-plugin</artifactId>
                    <version>2.10</version>
                    <configuration>
                        <skip>true</skip>
                    </configuration>
                </plugin>
            </plugins>
        </build>
    </project>

2.FlumeHbaseEventSerializer实现类代码如下：

package com.badou;
    
    import com.google.common.base.Charsets;
    import com.google.common.collect.Lists;
    import org.apache.commons.lang.RandomStringUtils;
    import org.apache.flume.Context;
    import org.apache.flume.Event;
    import org.apache.flume.FlumeException;
    import org.apache.flume.conf.ComponentConfiguration;
    import org.apache.flume.sink.hbase.HbaseEventSerializer;
    import org.apache.hadoop.hbase.client.Increment;
    import org.apache.hadoop.hbase.client.Put;
    import org.apache.hadoop.hbase.client.Row;
    import java.nio.charset.Charset;
    import java.util.ArrayList;
    import java.util.Calendar;
    import java.util.List;
    import java.util.concurrent.atomic.AtomicInteger;
    import java.util.regex.Pattern;
    
    public class FlumeHbaseEventSerializer implements HbaseEventSerializer {
        public static final String REGEX_CONFIG = "regex";
        public static final String REGEX_DEFAULT = " ";
        public static final String INGNORE_CASE_CONFIG = "regexIgnoreCase";
        public static final boolean INGNORE_CASE_DEFAULT = false;
        public static final String COL_NAME_CONFIG = "colNames";
        public static final String COL_NAME_DEFAULT = "ip";
        public static final String ROW_KEY_INDEX_CONFIG = "rowKeyIndex";
        public static final String ROW_KEY_NAME = "ROW_KEY";
        public static final String DEPOSIT_HEADERS_CONFIG = "depositHeaders";
        public static final boolean DEPOSIT_HEADERS_DEFAULT = false;
        public static final String CHARSET_CONFIG = "charset";
        public static final String CHARSET_DEFAULT = "utf-8";
    
        protected static final AtomicInteger nonce = new AtomicInteger(0);
        protected static String randomKey = RandomStringUtils.randomAlphanumeric(10);
        protected byte[]cf;
        private byte[]payload;
        private List<byte[]> colNames = Lists.newArrayList();
        private boolean regexIngnoreCase;
        private Charset charset;
    
    
        public void initialize(Event event, byte[] columnFamily) {
            event.getHeaders();
            this.payload = event.getBody();
            this.cf = columnFamily;
        }
    
        public List<Row> getActions() {
            ArrayList<Row> actions = Lists.newArrayList();
            byte[] rowKey;
            String body = new String(payload,charset);
            String tmp = body.replace("\"","");
            String[] arr = tmp.trim().split(" ");
            int len = arr.length;
            System.out.println("tmp:"+tmp);
            System.out.println("len1:"+len);
            String log_data = arr[len-1];
            String[] param_arr = log_data.split("&");
            System.out.println("===================="+log_data);
            String userid = param_arr[0].split("=")[1];
            String itemid = param_arr[1].split("=")[1];
            String type = param_arr[2].split("=")[1];
            String ip_str = param_arr[3].split("=")[1];
            System.out.println("=====================");
            System.out.println("=====================");
            System.out.println("=====================");
            System.out.println("=====================");
            System.out.println(userid);
            System.out.println(itemid);
            System.out.println(type);
            System.out.println(ip_str);
            System.out.println("=====================");
            System.out.println("=====================");
            System.out.println("=====================");
            System.out.println("=====================");
            try {
                System.out.println("0000=============");
                rowKey = getRowKey();
                System.out.println("1111=============");
                Put put = new Put(rowKey);
                put.add(cf,colNames.get(0),userid.getBytes(Charsets.UTF_8));
                put.add(cf,colNames.get(1),itemid.getBytes(Charsets.UTF_8));
                put.add(cf,colNames.get(2),type.getBytes(Charsets.UTF_8));
                put.add(cf,colNames.get(3),ip_str.getBytes(Charsets.UTF_8));
                actions.add(put);
            } catch (Exception e){
                throw new FlumeException("could not get row key!",e);
            }
            return actions;
        }
    
        public List<Increment> getIncrements() {
            return Lists.newArrayList();
        }
    
        public void close() {
    
        }
    
        public void configure(Context context) {
            String regex = context.getString(REGEX_CONFIG,REGEX_DEFAULT);
            regexIngnoreCase = context.getBoolean(INGNORE_CASE_CONFIG,INGNORE_CASE_DEFAULT);
            context.getBoolean(DEPOSIT_HEADERS_CONFIG,DEPOSIT_HEADERS_DEFAULT);
            Pattern.compile(regex, Pattern.DOTALL+(regexIngnoreCase? Pattern.CASE_INSENSITIVE: Pattern.UNIX_LINES));
            charset = Charset.forName(context.getString(CHARSET_CONFIG,CHARSET_DEFAULT));
    
            String cols = new String(context.getString("columns"));
            String colNameStr = "";
            if(cols != null && !"".equals(cols)){
                colNameStr = cols;
            }else {
                colNameStr = context.getString(COL_NAME_CONFIG,COL_NAME_DEFAULT);
            }
            String[]colmnNames = colNameStr.split(",");
            for (String s:colmnNames){
                colNames.add(s.getBytes(charset));
            }
        }
    
        public void configure(ComponentConfiguration componentConfiguration) {
    
        }
    
        protected byte[] getRowKey(Calendar cal){
            String str = new String(payload,charset);
            String tmp = str.replace("\"","");
            String[]arr = tmp.split(" ");
            int len = arr.length;
            System.out.println("tmp:"+tmp);
            System.out.println("len2:"+len);
            String log_data = arr[len-1];
            String[] param_arr = log_data.split("&");
            String userid = param_arr[0];
            String itemid = param_arr[1];
            String type = param_arr[2];
            String ip_str = param_arr[3];
            String rowKey = ip_str + "-"+ nonce.getAndIncrement();
            return rowKey.getBytes(charset);
        }
    
        protected byte[] getRowKey(){
            return getRowKey(Calendar.getInstance());
        }
    }

4.打jar包，直接在idea工具上双击clean，再双击install就出来了。

##### 5.用来连接flume和hbase的jar开发完成后，讲jar文件copy到flume主目录的lib目录下 #####

##### 6.开发cgi demo，接受NGINX转发过来的日志，并写入日志文件，供flume采集 #####

1.client.cpp代码如下：

using namespace std;
    
    
    inline void send_response(
                FCGX_Request& request, const string& resp_str) {
    
        FCGX_FPrintF(request.out, "Content-type: text/html;charset=utf-8\r\n\r\n");
        FCGX_FPrintF(request.out, "%s", resp_str.c_str());
    
        FCGX_Finish_r(&request);
    }
    
    int main(int argc, char **argv) {
    
        FCGX_Init();
        FCGX_Request request;
        FCGX_InitRequest(&request, 0, 0);
    
        FLAGS_log_dir = "/usr/local/logserver_study/cgi_demo/logs";
        FLAGS_max_log_size = 100;
        FLAGS_logbufsecs = 0;
        google::InitGoogleLogging(argv[0]);
    
        while(FCGX_Accept_r(&request) >= 0) {
            string query_str = FCGX_GetParam("QUERY_STRING", request.envp);
    
            cout << "query str: " << query_str << endl;
            LOG(INFO) << query_str;
            LOG(WARNING) << query_str;
            LOG(ERROR) << query_str;
    
            send_response(request, query_str);
        }
    
        return 0;
    }

2.代码编译：  
\]\# g++ -lglog -lfcgi client.cpp -o client  
3.加入托管：  
/usr/local/bin/spawn-fcgi -a 127.0.0.1 -p 8099 -f /usr/local/logserver\_study/cgi\_demo/client

##### 7.开发完成，开启服务，准备测试 #####

流程总结：  
1.启动Hadoop  
2.启动zookeeper  
3.启动hbase  
4.启动NGINX  
5.用spawn-fcgi工具托管自主开发的cgi demo  
6.启动flume的server端服务  
7.启动flume的客户端服务  
9.压力测试：ab -c 20 -n 5000 ‘http://slave2/badou\_recsys?userid=333&itemid=222&type=click&ip=1.0.0.10’

##### 8.hbase测试结果检查： #####

1.slave1和master都有日志文件产生  
2.检查hbase表：  
hbase(main):037:0> count ‘user\_action\_table’  
524 row(s) in 0.2830 seconds  
hbase数据插入成功

[format_png]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8xODE1NDM5MS04N2VhNjc0YTcxOWM3YzZjLnBuZw?x-oss-process=image/format,png
[format_png 1]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8xODE1NDM5MS00NGM0NzIyZDA0OTIzYWVmLnBuZw?x-oss-process=image/format,png
[format_png 2]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8xODE1NDM5MS03MmI4OTY0MGY5NzgxNzY0LnBuZw?x-oss-process=image/format,png