LOADING STUFF...

Apache Phoenix Flume集成 JsonEventSerializer改进

–>

背景

数据从kafka ingest到Phoenix。数据格式采取Json。数据链路:

api -> kafka -> Flume -> Phoenix

官方JsonEventSerializer的问题

  • 每个table column必须有json字段,如果某个字段json中没有,那么这条记录被丢弃;
    例如table中有cola和colb两个列,但是json数据中只有{“cola”:1},原生版本会将这个消息丢弃。

  • 数组的问题,json中的数组元素是没有类型的,例如一个小数字会被解析为java的int,如果我们的表中数组元素定义为BITINT,此时会有integer转long类型的exception。

  • timestamp不支持unix timestamp格式(long)

  • 如果数据格式出错会有SqlException异常抛出,此时会导致Flume sink循环拉取错误的消息并不停地尝试插入数据到Phoenix,但是一直失败,新的数据也无法插入。

内部原理

  • 读取flume的event body (String),利用JsonObject解析,这里的处理会将json的每个字段强转为String类型;
  • 如果该json字段对应的table列类型是isArrayType则将json值(数组)创建成Phoenix Array;
  • 如果不是数组,则直接将json的值转为对应的Object upsertValue = pDataType.toObject(jsonValue)
  • 最后调用Phoenix jdbc API (PrepareedStatement.excute())将数据插入到Phoenix,这里会有类型判断。

解决:

  • 为null字段调用setNull
  • 把json数组元素逐个转为table中元素的类型
  • 对timestamp类型特殊处理,判断数据格式是不是\d+,如果是,则强转
else if (pDataType == PTimestamp.INSTANCE) {
          if (value.matches("\\d+")) { // if it's  a Long value as time stamp
                  upsertValue = pDataType.toObject(Long.parseLong(value), PLong.INSTANCE);
          } else {
                  upsertValue = pDataType.toObject(value);
          }
  }
  • 避免抛出SqlException,只打印错误消息。

代码修改参见:https://github.com/wlu-mstr/phoenix/tree/4.13-cdh5.11.2

原文链接:https://www.cnblogs.com/luweiseu/p/8872826.html
本文来源 互联网收集,文章内容系作者个人观点,不代表 本站 对观点赞同或支持。如需转载,请注明文章来源,如您发现有涉嫌抄袭侵权的内容,请联系本站核实处理。

© 版权声明

相关文章