189 8069 5689
这篇文章主要介绍“Flume怎么自定义Event Serializer序列化类”，在日常操作中，相信很多人在Flume怎么自定义Event Serializer序列化类问题上存在疑惑，小编查阅了各式资料，整理出简单好用的操作方法，希望对大家解答”Flume怎么自定义Event Serializer序列化类”的疑惑有所帮助！接下来，请跟着小编一起来学习吧！
向阳网站建设公司创新互联建站,向阳网站设计制作，有大型网站制作公司丰富经验。已为向阳1000多家提供企业网站建设服务。企业网站搭建\成都外贸网站建设要多少钱，请找那个售后服务好的向阳做网站的公司定做！
把日志从flume打到hbase中，但是我们的日志由于前期是存到MongoDB中的，所以都是Json格式的日志，这时候使用flume自带的SimpleHbaseEventSerializer和RegexHbaseEventSerializer这样的就不行了，于是开始痛苦的看源码，自己写序列化的类（这里需要注意，如果是在flume的hbasesink包下编写的代码，License信息一定要加上。就是最上面那段英文，要不然在运行的时候会报错），比较简单，编写好类之后，编译打包，传到flume的lib目录下，然后在配置agent的时候指定Serializer的类为编写的类即可。下面是代码（类注释没贴出来，见谅哈）：
public class PRTMSAsyncHbaseEventSerializer implements AsyncHbaseEventSerializer {
private byte[] table;//hbase表
private byte[] cf;//列簇
private byte[][] payload;//列集合
private byte[][] payloadColumn;//列值
private byte[] incrementColumn;
private String rowSuffix;//roykey后缀
private String rowPrefix;//rowkey前缀
private byte[] incrementRow;
private KeyType keyType;//rowkey后缀类型
private static final Logger logger = LoggerFactory.getLogger(PRTMSAsyncHbaseEventSerializer.class);
@Override
public void configure(Context context) {
// TODO Auto-generated method stub
//设置主键后缀类型，这里使用时间戳
keyType = KeyType.TS;
if (iCol != null && !iCol.isEmpty()) {
incrementColumn = iCol.getBytes(Charsets.UTF_8);
}
incrementRow = context.getString("incrementRow", "incRow").getBytes(Charsets.UTF_8);
}
@Override
public void configure(ComponentConfiguration conf) {
// TODO Auto-generated method stub
}
@Override
public void initialize(byte[] table, byte[] cf) {
// TODO Auto-generated method stub
this.table = table;
this.cf = cf;
}
/**
*
* @Title: setEvent
* @Description: 获取日志信息，并解析出HBase的列以及列的value值
* @param event
* @throws
* @see org.apache.flume.sink.hbase.AsyncHbaseEventSerializer#setEvent(org.apache.flume.Event)
*/
@Override
public void setEvent(Event event) {
// TODO Auto-generated method stub
//获取日志信息
String log = new String(event.getBody(), StandardCharsets.UTF_8);
//headers包含日志中项目编号和host信息
Map headers = event.getHeaders();
JsonReader jsonReader = new JsonReader(new StringReader(log));
String name = "";
String value = "";
String path = "";
Map kv = new HashMap();
try {
//解析日志中的键值对缓存到map中
jsonReader.beginObject();
while (jsonReader.hasNext()) {
name = jsonReader.nextName();
value = jsonReader.nextString();
if(name.equals("uri"))
path = value.split(" ")[1];
kv.put(name, value);
}
jsonReader.endObject();
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
//解析headers中的项目id和服务host、路径
if(path.contains("?")){
path = path.substring(0, path.indexOf("?"));
}
String pcode = headers.get("pcode");
String host = headers.get("host");
//将项目编号和服务器host添加到map中
kv.put("pcode",pcode);
kv.put("host", host);
//初始化列和value数组
this.payloadColumn = new byte[kv.keySet().size()][];
this.payload = new byte[kv.keySet().size()][];
int i = 0;
//给hbase的列和value赋值
for (String key : kv.keySet()) {
this.payloadColumn[i] = key.getBytes();
this.payload[i] = kv.get(key).getBytes();
i++;
}
//设置rowkey的前缀 格式是项目编号+路径
this.rowSuffix = new StringBuilder(pcode).reverse().toString() + ":"+path+":"+kv.get("time");
}
@Override
public List getActions() {
// TODO Auto-generated method stub
List actions = new ArrayList();
if (payloadColumn != null) {
byte[] rowKey;
try {
rowKey = rowSuffix.getBytes();
// for 循环，提交所有列和对于数据的put请求。
for (int i = 0; i < this.payload.length; i++) {
PutRequest putRequest = new PutRequest(table, rowKey, cf, payloadColumn[i], payload[i]);
actions.add(putRequest);
}
} catch (Exception e) {
throw new FlumeException("Could not get row key!", e);
}
}
return actions;
}
@Override
public List getIncrements() {
// TODO Auto-generated method stub
List actions = new ArrayList();
if (incrementColumn != null) {
AtomicIncrementRequest inc = new AtomicIncrementRequest(table, incrementRow, cf, incrementColumn);
actions.add(inc);
}
return actions;
}
@Override
public void cleanUp() {
// TODO Auto-generated method stub
}
} 到此，关于“Flume怎么自定义Event Serializer序列化类”的学习就结束了，希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习，快去试试吧！若想继续学习更多相关知识，请继续关注创新互联网站，小编会继续努力为大家带来更多实用的文章！