hadoop - 无法使用Apache Flume将数据插入HBase

Question

我正在使用flume将日志文件数据插入到hbase table。但是没有任何内容插入到表中。
水槽剂如下:

 agent1.sources = tail
 agent1.channels = memoryChannel
 agent1.sinks = loggerSink sink1
 agent1.sources.tail.type = exec
agent1.sources.tail.command = tail -f /usr/local/jarsfortest/LogsForTest/generatingLogs-app.logs
agent1.sources.tail.channels = memoryChannel

agent1.sinks.loggerSink.channel = memoryChannel
agent1.sinks.loggerSink.type = logger

agent1.sinks.sink1.type = org.apache.flume.sink.hbase.HBaseSink
agent1.sinks.sink1.channel = memoryChannel
agent1.sinks.sink1.table = testFlume
agent1.sinks.sink1.columnFamily = log
agent1.sinks.sink1.serializer = org.apache.flume.sink.hbase.RegexHbaseEventSerializer
#agent1.sinks.sink1.serializer.regex = [a-zA-Z0-9]*[^C][a-zA-Z0-9]*[^C][a-zA-Z0-9]*
agent1.sinks.sink1.serializer.regex =[a-zA-Z0-9]*^C[a-zA-Z0-9]*^C[a-zA-Z0-9]*
agent1.sinks.sink1.serializer.colNames = id, no_fill_reason, bid

agent1.channels.memoryChannel.type = memory
agent1.channels.memoryChannel.capacity = 1000

以上代理已成功启动..但日志文件数据未插入到hbase中。
日志文件数据如下:
id0 ^ COK ^ C10
即数据是由控制字符分隔的。
请帮我。
提前致谢。

Answer 1

原因可能是正则表达式不匹配
调试此请按照以下步骤

1)使用此选项只能启动一个代理

flume-ng agent -n $1 -c ../../config/conf/ -f ../../config/conf/$1.conf -Xmx3g **-Xdebug -Xrunjdwp:transport=dt_socket, address=1044,server=y,suspend=y**
  --classpath ../lib/*:../../config/conf/zoo.cfg:../.

一旦启动脚本

listening to 1044...

消息会来。

2)创建eclipse远程应用程序以连接到您的服务器名称和端口为1044

3) getActions方法负责将行放入hbase。

在 getActions的 EventSerializer方法中放置一个断点

public List<Row> getActions() throws FlumeException {
        List actions = Lists.newArrayList();

        Matcher m = this.inputPattern.matcher(new String(this.payload, this.charset));
        if (!(m.matches())) {
            return Lists.newArrayList();
        }

        if (m.groupCount() != this.colNames.size())
            return Lists.newArrayList();
        try {
            byte[] rowKey;
            byte[] rowKey;
            if (this.rowKeyIndex < 0)
                rowKey = getRowKey();
            else {
                rowKey = m.group(this.rowKeyIndex + 1).getBytes(Charsets.UTF_8);
            }
            Put put = new Put(rowKey);

            for (int i = 0; i < this.colNames.size(); ++i) {
                if (i != this.rowKeyIndex) {
                    put.add(this.cf, (byte[]) this.colNames.get(i), m.group(i + 1).getBytes(Charsets.UTF_8));
                }
            }
            if (this.depositHeaders) {
                for (Map.Entry entry : this.headers.entrySet()) {
                    put.add(this.cf, ((String) entry.getKey()).getBytes(this.charset),
                                    ((String) entry.getValue()).getBytes(this.charset));
                }
            }
            actions.add(put);
        } catch (Exception e) {
            throw new FlumeException("Could not get row key!", e);
        }
        return actions;
    }