分享

flume学习(六):使用hive来分析flume收集的日志数据

本帖最后由 坎蒂丝_Swan 于 2015-3-19 22:08 编辑


问题导读

1.遇到无法转换成JSON对象的字符串时应如何处理?
2.遇到非JSON格式输入的时候应如何处理?







前面已经讲过如何将log4j的日志输出到指定的hdfs目录,我们前面的指定目录为/flume/events。
如果想用hive来分析采集来的日志,我们可以将/flume/events下面的日志数据都load到hive中的表当中去。

如果了解hive的load data原理的话,还有一种更简便的方式,可以省去load data这一步,就是直接将sink1.hdfs.path指定为hive表的目录。
下面我将详细描述具体的操作步骤。

我们还是从需求驱动来讲解,前面我们采集的数据,都是接口的访问日志数据,数据格式是JSON格式如下:

{"requestTime":1405651379758,"requestParams":{"timestamp":1405651377211,"phone":"02038824941","cardName":"测试商家名称","provinceCode":"440000","cityCode":"440106"},"requestUrl":"/reporter-api/reporter/reporter12/init.do"}

现在有一个需求,我们要统计接口的总调用量。

我第一想法就是,hive中建一张表:test             然后将hdfs.path指定为tier1.sinks.sink1.hdfs.path=hdfs://master68:8020/user/hive/warehouse/besttone.db/test

然后select  count(*) from test;   完事。
这个方案简单,粗暴,先这么干着。于是会遇到一个问题,我的日志数据时JSON格式的,需要hive来序列化和反序列化JSON格式的数据到test表的具体字段当中去。

这有点糟糕,因为hive本身没有提供JSON的SERDE,但是有提供函数来解析JSON字符串,

第一个是(UDF):
get_json_object(string json_string,string path) 从给定路径上的JSON字符串中抽取出JSON对象,并返回这个对象的JSON字符串形式,如果输入的JSON字符串是非法的,则返回NULL。

第二个是表生成函数(UDTF):json_tuple(string jsonstr,p1,p2,...,pn) 本函数可以接受多个标签名称,对输入的JSON字符串进行处理,这个和get_json_object这个UDF类似,不过更高效,其通过一次调用就可以获得多个键值,例:select b.* from test_json a lateral view json_tuple(a.id,'id','name') b as f1,f2;通过lateral view行转列。

最理想的方式就是能有一种JSON SERDE,只要我们LOAD完数据,就直接可以select * from test,而不是select get_json_object这种方式来获取,N个字段就要解析N次,效率太低了。

好在cloudrea wiki里提供了一个json serde类(这个类没有在发行的hive的jar包中),于是我把它搬来了,如下:
  1. package com.besttone.hive.serde;
  2. import java.util.ArrayList;
  3. import java.util.Arrays;
  4. import java.util.HashMap;
  5. import java.util.List;
  6. import java.util.Map;
  7. import java.util.Properties;
  8. import org.apache.hadoop.conf.Configuration;
  9. import org.apache.hadoop.hive.serde.serdeConstants;
  10. import org.apache.hadoop.hive.serde2.SerDe;
  11. import org.apache.hadoop.hive.serde2.SerDeException;
  12. import org.apache.hadoop.hive.serde2.SerDeStats;
  13. import org.apache.hadoop.hive.serde2.objectinspector.ListObjectInspector;
  14. import org.apache.hadoop.hive.serde2.objectinspector.MapObjectInspector;
  15. import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
  16. import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;
  17. import org.apache.hadoop.hive.serde2.objectinspector.StructField;
  18. import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
  19. import org.apache.hadoop.hive.serde2.typeinfo.ListTypeInfo;
  20. import org.apache.hadoop.hive.serde2.typeinfo.MapTypeInfo;
  21. import org.apache.hadoop.hive.serde2.typeinfo.StructTypeInfo;
  22. import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
  23. import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory;
  24. import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils;
  25. import org.apache.hadoop.io.Text;
  26. import org.apache.hadoop.io.Writable;
  27. import org.codehaus.jackson.map.ObjectMapper;
  28. /**
  29. * This SerDe can be used for processing JSON data in Hive. It supports
  30. * arbitrary JSON data, and can handle all Hive types except for UNION. However,
  31. * the JSON data is expected to be a series of discrete records, rather than a
  32. * JSON array of objects.
  33. *
  34. * The Hive table is expected to contain columns with names corresponding to
  35. * fields in the JSON data, but it is not necessary for every JSON field to have
  36. * a corresponding Hive column. Those JSON fields will be ignored during
  37. * queries.
  38. *
  39. * Example:
  40. *
  41. * { "a": 1, "b": [ "str1", "str2" ], "c": { "field1": "val1" } }
  42. *
  43. * Could correspond to a table:
  44. *
  45. * CREATE TABLE foo (a INT, b ARRAY<STRING>, c STRUCT<field1:STRING>);
  46. *
  47. * JSON objects can also interpreted as a Hive MAP type, so long as the keys and
  48. * values in the JSON object are all of the appropriate types. For example, in
  49. * the JSON above, another valid table declaraction would be:
  50. *
  51. * CREATE TABLE foo (a INT, b ARRAY<STRING>, c MAP<STRING,STRING>);
  52. *
  53. * Only STRING keys are supported for Hive MAPs.
  54. */
  55. public class JSONSerDe implements SerDe {
  56.         private StructTypeInfo rowTypeInfo;
  57.         private ObjectInspector rowOI;
  58.         private List<String> colNames;
  59.         private List<Object> row = new ArrayList<Object>();
  60.         //遇到非JSON格式输入的时候的处理。
  61.         private boolean ignoreInvalidInput;
  62.         /**
  63.          * An initialization function used to gather information about the table.
  64.          * Typically, a SerDe implementation will be interested in the list of
  65.          * column names and their types. That information will be used to help
  66.          * perform actual serialization and deserialization of data.
  67.          */
  68.         @Override
  69.         public void initialize(Configuration conf, Properties tbl)
  70.                         throws SerDeException {
  71.                 // 遇到无法转换成JSON对象的字符串时,是否忽略,默认不忽略,抛出异常,设置为true将跳过异常。
  72.                 ignoreInvalidInput = Boolean.valueOf(tbl.getProperty(
  73.                                 "input.invalid.ignore", "false"));
  74.                 // Get a list of the table's column names.
  75.                 String colNamesStr = tbl.getProperty(serdeConstants.LIST_COLUMNS);
  76.                 colNames = Arrays.asList(colNamesStr.split(","));
  77.                 // Get a list of TypeInfos for the columns. This list lines up with
  78.                 // the list of column names.
  79.                 String colTypesStr = tbl.getProperty(serdeConstants.LIST_COLUMN_TYPES);
  80.                 List<TypeInfo> colTypes = TypeInfoUtils
  81.                                 .getTypeInfosFromTypeString(colTypesStr);
  82.                 rowTypeInfo = (StructTypeInfo) TypeInfoFactory.getStructTypeInfo(
  83.                                 colNames, colTypes);
  84.                 rowOI = TypeInfoUtils
  85.                                 .getStandardJavaObjectInspectorFromTypeInfo(rowTypeInfo);
  86.         }
  87.         /**
  88.          * This method does the work of deserializing a record into Java objects
  89.          * that Hive can work with via the ObjectInspector interface. For this
  90.          * SerDe, the blob that is passed in is a JSON string, and the Jackson JSON
  91.          * parser is being used to translate the string into Java objects.
  92.          *
  93.          * The JSON deserialization works by taking the column names in the Hive
  94.          * table, and looking up those fields in the parsed JSON object. If the
  95.          * value of the field is not a primitive, the object is parsed further.
  96.          */
  97.         @Override
  98.         public Object deserialize(Writable blob) throws SerDeException {
  99.                 Map<?, ?> root = null;
  100.                 row.clear();
  101.                 try {
  102.                         ObjectMapper mapper = new ObjectMapper();
  103.                         // This is really a Map<String, Object>. For more information about
  104.                         // how
  105.                         // Jackson parses JSON in this example, see
  106.                         // http://wiki.fasterxml.com/JacksonDataBinding
  107.                         root = mapper.readValue(blob.toString(), Map.class);
  108.                 } catch (Exception e) {
  109.                         // 如果为true,不抛出异常,忽略该行数据
  110.                         if (!ignoreInvalidInput)
  111.                                 throw new SerDeException(e);
  112.                         else {
  113.                                 return null;
  114.                         }
  115.                         
  116.                 }
  117.                 // Lowercase the keys as expected by hive
  118.                 Map<String, Object> lowerRoot = new HashMap();
  119.                 for (Map.Entry entry : root.entrySet()) {
  120.                         lowerRoot.put(((String) entry.getKey()).toLowerCase(),
  121.                                         entry.getValue());
  122.                 }
  123.                 root = lowerRoot;
  124.                 Object value = null;
  125.                 for (String fieldName : rowTypeInfo.getAllStructFieldNames()) {
  126.                         try {
  127.                                 TypeInfo fieldTypeInfo = rowTypeInfo
  128.                                                 .getStructFieldTypeInfo(fieldName);
  129.                                 value = parseField(root.get(fieldName), fieldTypeInfo);
  130.                         } catch (Exception e) {
  131.                                 value = null;
  132.                         }
  133.                         row.add(value);
  134.                 }
  135.                 return row;
  136.         }
  137.         /**
  138.          * Parses a JSON object according to the Hive column's type.
  139.          *
  140.          * @param field
  141.          *            - The JSON object to parse
  142.          * @param fieldTypeInfo
  143.          *            - Metadata about the Hive column
  144.          * @return - The parsed value of the field
  145.          */
  146.         private Object parseField(Object field, TypeInfo fieldTypeInfo) {
  147.                 switch (fieldTypeInfo.getCategory()) {
  148.                 case PRIMITIVE:
  149.                         // Jackson will return the right thing in this case, so just return
  150.                         // the object
  151.                         if (field instanceof String) {
  152.                                 field = field.toString().replaceAll("\n", "\\\\n");
  153.                         }
  154.                         return field;
  155.                 case LIST:
  156.                         return parseList(field, (ListTypeInfo) fieldTypeInfo);
  157.                 case MAP:
  158.                         return parseMap(field, (MapTypeInfo) fieldTypeInfo);
  159.                 case STRUCT:
  160.                         return parseStruct(field, (StructTypeInfo) fieldTypeInfo);
  161.                 case UNION:
  162.                         // Unsupported by JSON
  163.                 default:
  164.                         return null;
  165.                 }
  166.         }
  167.         /**
  168.          * Parses a JSON object and its fields. The Hive metadata is used to
  169.          * determine how to parse the object fields.
  170.          *
  171.          * @param field
  172.          *            - The JSON object to parse
  173.          * @param fieldTypeInfo
  174.          *            - Metadata about the Hive column
  175.          * @return - A map representing the object and its fields
  176.          */
  177.         private Object parseStruct(Object field, StructTypeInfo fieldTypeInfo) {
  178.                 Map<Object, Object> map = (Map<Object, Object>) field;
  179.                 ArrayList<TypeInfo> structTypes = fieldTypeInfo
  180.                                 .getAllStructFieldTypeInfos();
  181.                 ArrayList<String> structNames = fieldTypeInfo.getAllStructFieldNames();
  182.                 List<Object> structRow = new ArrayList<Object>(structTypes.size());
  183.                 for (int i = 0; i < structNames.size(); i++) {
  184.                         structRow.add(parseField(map.get(structNames.get(i)),
  185.                                         structTypes.get(i)));
  186.                 }
  187.                 return structRow;
  188.         }
  189.         /**
  190.          * Parse a JSON list and its elements. This uses the Hive metadata for the
  191.          * list elements to determine how to parse the elements.
  192.          *
  193.          * @param field
  194.          *            - The JSON list to parse
  195.          * @param fieldTypeInfo
  196.          *            - Metadata about the Hive column
  197.          * @return - A list of the parsed elements
  198.          */
  199.         private Object parseList(Object field, ListTypeInfo fieldTypeInfo) {
  200.                 ArrayList<Object> list = (ArrayList<Object>) field;
  201.                 TypeInfo elemTypeInfo = fieldTypeInfo.getListElementTypeInfo();
  202.                 for (int i = 0; i < list.size(); i++) {
  203.                         list.set(i, parseField(list.get(i), elemTypeInfo));
  204.                 }
  205.                 return list.toArray();
  206.         }
  207.         /**
  208.          * Parse a JSON object as a map. This uses the Hive metadata for the map
  209.          * values to determine how to parse the values. The map is assumed to have a
  210.          * string for a key.
  211.          *
  212.          * @param field
  213.          *            - The JSON list to parse
  214.          * @param fieldTypeInfo
  215.          *            - Metadata about the Hive column
  216.          * @return
  217.          */
  218.         private Object parseMap(Object field, MapTypeInfo fieldTypeInfo) {
  219.                 Map<Object, Object> map = (Map<Object, Object>) field;
  220.                 TypeInfo valueTypeInfo = fieldTypeInfo.getMapValueTypeInfo();
  221.                 for (Map.Entry<Object, Object> entry : map.entrySet()) {
  222.                         map.put(entry.getKey(), parseField(entry.getValue(), valueTypeInfo));
  223.                 }
  224.                 return map;
  225.         }
  226.         /**
  227.          * Return an ObjectInspector for the row of data
  228.          */
  229.         @Override
  230.         public ObjectInspector getObjectInspector() throws SerDeException {
  231.                 return rowOI;
  232.         }
  233.         /**
  234.          * Unimplemented
  235.          */
  236.         @Override
  237.         public SerDeStats getSerDeStats() {
  238.                 return null;
  239.         }
  240.         /**
  241.          * JSON is just a textual representation, so our serialized class is just
  242.          * Text.
  243.          */
  244.         @Override
  245.         public Class<? extends Writable> getSerializedClass() {
  246.                 return Text.class;
  247.         }
  248.         /**
  249.          * This method takes an object representing a row of data from Hive, and
  250.          * uses the ObjectInspector to get the data for each column and serialize
  251.          * it. This implementation deparses the row into an object that Jackson can
  252.          * easily serialize into a JSON blob.
  253.          */
  254.         @Override
  255.         public Writable serialize(Object obj, ObjectInspector oi)
  256.                         throws SerDeException {
  257.                 Object deparsedObj = deparseRow(obj, oi);
  258.                 ObjectMapper mapper = new ObjectMapper();
  259.                 try {
  260.                         // Let Jackson do the work of serializing the object
  261.                         return new Text(mapper.writeValueAsString(deparsedObj));
  262.                 } catch (Exception e) {
  263.                         throw new SerDeException(e);
  264.                 }
  265.         }
  266.         /**
  267.          * Deparse a Hive object into a Jackson-serializable object. This uses the
  268.          * ObjectInspector to extract the column data.
  269.          *
  270.          * @param obj
  271.          *            - Hive object to deparse
  272.          * @param oi
  273.          *            - ObjectInspector for the object
  274.          * @return - A deparsed object
  275.          */
  276.         private Object deparseObject(Object obj, ObjectInspector oi) {
  277.                 switch (oi.getCategory()) {
  278.                 case LIST:
  279.                         return deparseList(obj, (ListObjectInspector) oi);
  280.                 case MAP:
  281.                         return deparseMap(obj, (MapObjectInspector) oi);
  282.                 case PRIMITIVE:
  283.                         return deparsePrimitive(obj, (PrimitiveObjectInspector) oi);
  284.                 case STRUCT:
  285.                         return deparseStruct(obj, (StructObjectInspector) oi, false);
  286.                 case UNION:
  287.                         // Unsupported by JSON
  288.                 default:
  289.                         return null;
  290.                 }
  291.         }
  292.         /**
  293.          * Deparses a row of data. We have to treat this one differently from other
  294.          * structs, because the field names for the root object do not match the
  295.          * column names for the Hive table.
  296.          *
  297.          * @param obj
  298.          *            - Object representing the top-level row
  299.          * @param structOI
  300.          *            - ObjectInspector for the row
  301.          * @return - A deparsed row of data
  302.          */
  303.         private Object deparseRow(Object obj, ObjectInspector structOI) {
  304.                 return deparseStruct(obj, (StructObjectInspector) structOI, true);
  305.         }
  306.         /**
  307.          * Deparses struct data into a serializable JSON object.
  308.          *
  309.          * @param obj
  310.          *            - Hive struct data
  311.          * @param structOI
  312.          *            - ObjectInspector for the struct
  313.          * @param isRow
  314.          *            - Whether or not this struct represents a top-level row
  315.          * @return - A deparsed struct
  316.          */
  317.         private Object deparseStruct(Object obj, StructObjectInspector structOI,
  318.                         boolean isRow) {
  319.                 Map<Object, Object> struct = new HashMap<Object, Object>();
  320.                 List<? extends StructField> fields = structOI.getAllStructFieldRefs();
  321.                 for (int i = 0; i < fields.size(); i++) {
  322.                         StructField field = fields.get(i);
  323.                         // The top-level row object is treated slightly differently from
  324.                         // other
  325.                         // structs, because the field names for the row do not correctly
  326.                         // reflect
  327.                         // the Hive column names. For lower-level structs, we can get the
  328.                         // field
  329.                         // name from the associated StructField object.
  330.                         String fieldName = isRow ? colNames.get(i) : field.getFieldName();
  331.                         ObjectInspector fieldOI = field.getFieldObjectInspector();
  332.                         Object fieldObj = structOI.getStructFieldData(obj, field);
  333.                         struct.put(fieldName, deparseObject(fieldObj, fieldOI));
  334.                 }
  335.                 return struct;
  336.         }
  337.         /**
  338.          * Deparses a primitive type.
  339.          *
  340.          * @param obj
  341.          *            - Hive object to deparse
  342.          * @param oi
  343.          *            - ObjectInspector for the object
  344.          * @return - A deparsed object
  345.          */
  346.         private Object deparsePrimitive(Object obj, PrimitiveObjectInspector primOI) {
  347.                 return primOI.getPrimitiveJavaObject(obj);
  348.         }
  349.         private Object deparseMap(Object obj, MapObjectInspector mapOI) {
  350.                 Map<Object, Object> map = new HashMap<Object, Object>();
  351.                 ObjectInspector mapValOI = mapOI.getMapValueObjectInspector();
  352.                 Map<?, ?> fields = mapOI.getMap(obj);
  353.                 for (Map.Entry<?, ?> field : fields.entrySet()) {
  354.                         Object fieldName = field.getKey();
  355.                         Object fieldObj = field.getValue();
  356.                         map.put(fieldName, deparseObject(fieldObj, mapValOI));
  357.                 }
  358.                 return map;
  359.         }
  360.         /**
  361.          * Deparses a list and its elements.
  362.          *
  363.          * @param obj
  364.          *            - Hive object to deparse
  365.          * @param oi
  366.          *            - ObjectInspector for the object
  367.          * @return - A deparsed object
  368.          */
  369.         private Object deparseList(Object obj, ListObjectInspector listOI) {
  370.                 List<Object> list = new ArrayList<Object>();
  371.                 List<?> field = listOI.getList(obj);
  372.                 ObjectInspector elemOI = listOI.getListElementObjectInspector();
  373.                 for (Object elem : field) {
  374.                         list.add(deparseObject(elem, elemOI));
  375.                 }
  376.                 return list;
  377.         }
  378. }
复制代码
我稍微修改了一点东西,多加了一个参数input.invalid.ignore,对应的变量为:


//遇到非JSON格式输入的时候的处理。
private boolean ignoreInvalidInput;

在deserialize方法中原来是如果传入的是非JSON格式字符串的话,直接抛出了SerDeException,我加了一个参数来控制它是否抛出异常,在initialize方法中初始化这个变量(默认为false):

// 遇到无法转换成JSON对象的字符串时,是否忽略,默认不忽略,抛出异常,设置为true将跳过异常。
ignoreInvalidInput = Boolean.valueOf(tbl.getProperty(
"input.invalid.ignore", "false"));

好的,现在将这个类打成JAR包: JSONSerDe.jar,放在hive_home的auxlib目录下(我的是/etc/hive/auxlib),然后修改hive-env.sh,添加HIVE_AUX_JARS_PATH=/etc/hive/auxlib/JSONSerDe.jar,这样每次运行hive客户端的时候都会将这个jar包添加到classpath,否则在设置SERDE的时候会报找不到类。

现在我们在HIVE中创建一张表用来存放日志数据:
  1. create table test(
  2. requestTime BIGINT,
  3. requestParams STRUCT<timestamp:BIGINT,phone:STRING,cardName:STRING,provinceCode:STRING,cityCode:STRING>,        
  4. requestUrl STRING)
  5. row format serde "com.besttone.hive.serde.JSONSerDe"
  6. WITH SERDEPROPERTIES(
  7. "input.invalid.ignore"="true",
  8. "requestTime"="$.requestTime",
  9. "requestParams.timestamp"="$.requestParams.timestamp",
  10. "requestParams.phone"="$.requestParams.phone",
  11. "requestParams.cardName"="$.requestParams.cardName",
  12. "requestParams.provinceCode"="$.requestParams.provinceCode",
  13. "requestParams.cityCode"="$.requestParams.cityCode",
  14. "requestUrl"="$.requestUrl");
复制代码
这个表结构就是按照日志格式设计的,还记得前面说过的日志数据如下:

{"requestTime":1405651379758,"requestParams":{"timestamp":1405651377211,"phone":"02038824941","cardName":"测试商家名称","provinceCode":"440000","cityCode":"440106"},"requestUrl":"/reporter-api/reporter/reporter12/init.do"}

我使用了一个STRUCT类型来保存requestParams的值,row format我们用的是自定义的json serde:com.besttone.hive.serde.JSONSerDe,SERDEPROPERTIES中,除了设置JSON对象的映射关系外,我还设置了一个自定义的参数:"input.invalid.ignore"="true",忽略掉所有非JSON格式的输入行。

这里不是真正意义的忽略,只是非法行的每个输出字段都为NULL了,要在结果集上忽略,必须这样写:select * from test where requestUrl is not null;
OK表建好了,现在就差数据了,我们启动flumedemo的WriteLog,往hive表test目录下面输出一些日志数据,然后在进入hive客户端,select * from test;所以字段都正确的解析,大功告成。

flume.conf如下:
  1. tier1.sources=source1
  2. tier1.channels=channel1
  3. tier1.sinks=sink1
  4. tier1.sources.source1.type=avro
  5. tier1.sources.source1.bind=0.0.0.0
  6. tier1.sources.source1.port=44444
  7. tier1.sources.source1.channels=channel1
  8. tier1.sources.source1.interceptors=i1 i2
  9. tier1.sources.source1.interceptors.i1.type=regex_filter
  10. tier1.sources.source1.interceptors.i1.regex=\\{.*\\}
  11. tier1.sources.source1.interceptors.i2.type=timestamp
  12. tier1.channels.channel1.type=memory
  13. tier1.channels.channel1.capacity=10000
  14. tier1.channels.channel1.transactionCapacity=1000
  15. tier1.channels.channel1.keep-alive=30
  16. tier1.sinks.sink1.type=hdfs
  17. tier1.sinks.sink1.channel=channel1
  18. tier1.sinks.sink1.hdfs.path=hdfs://master68:8020/user/hive/warehouse/besttone.db/test
  19. tier1.sinks.sink1.hdfs.fileType=DataStream
  20. tier1.sinks.sink1.hdfs.writeFormat=Text
  21. tier1.sinks.sink1.hdfs.rollInterval=0
  22. tier1.sinks.sink1.hdfs.rollSize=10240
  23. tier1.sinks.sink1.hdfs.rollCount=0
  24. tier1.sinks.sink1.hdfs.idleTimeout=60
复制代码
besttone.db是我在hive中创建的数据库,了解hive的应该理解没多大问题。

OK,到这篇文章为止,整个从LOG4J生产日志,到flume收集日志,再到用hive离线分析日志,一整套流水线都讲解完了。



flume学习(五):flume将log4j日志数据写入到hdfs
flume学习(六):使用hive来分析flume收集的日志数据
flume学习(七)、(八):如何使用event header中的key值以及自定义source
flume学习(九):自定义拦截器
flume学习(十):使用Morphline Interceptor
flume学习(十一):如何使用Spooling Directory Source
欢迎加入about云群90371779322273151432264021 ,云计算爱好者群,亦可关注about云腾讯认证空间||关注本站微信

已有(1)人评论

跳转到指定楼层
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

关闭

推荐上一条 /2 下一条