给一个2层嵌套的数据
{"metric":"smsSendSucessCout1","nested":{"clientId":"client-id","number":20},"time":1537075089042,"value":"1.0"}
这个其实只要定义好schema就可以了，直接上源码
1/** 2 * 实体对象 3 * @author liu.zhiqiang 4 * @version $Id: AthenaxKafkaDTO.java, v 0.1 2018-09-12 18:14 liu.zhiqiang Exp $$ 5 */ 6@Data 7public class AthenaxKafkaDTO { 8 private String topic; 9 private String brokerAddress; 10 private String groupId; 11 private List<AthenaxJSONSchema> schemas; 12 13 // org.apache.flink.formats.json.JsonRowDeserializationSchema.deserialize(JsonRowDeserializationSchema.java:97) 14 private TypeInformation<?> getTypeInfo(AthenaxJSONSchema schema) throws TableNotExistException { 15 String type = schema.getType(); 16 if (type.toLowerCase().equals("string")) { 17 return BasicTypeInfo.STRING_TYPE_INFO; 18 } else if (type.toLowerCase().equals("boolean")) { 19 return BasicTypeInfo.BOOLEAN_TYPE_INFO; 20 } else if (type.toLowerCase().equals("byte")) { 21 return BasicTypeInfo.BYTE_TYPE_INFO; 22 } else if (type.toLowerCase().equals("short")) { 23 return BasicTypeInfo.SHORT_TYPE_INFO; 24 } else if (type.toLowerCase().equals("int")) { 25 return BasicTypeInfo.INT_TYPE_INFO; 26 } else if (type.toLowerCase().equals("long")) { 27 return BasicTypeInfo.LONG_TYPE_INFO; 28 } else if (type.toLowerCase().equals("float")) { 29 return BasicTypeInfo.FLOAT_TYPE_INFO; 30 } else if (type.toLowerCase().equals("double")) { 31 return BasicTypeInfo.DOUBLE_TYPE_INFO; 32 } else if (type.toLowerCase().equals("char")) { 33 return BasicTypeInfo.CHAR_TYPE_INFO; 34 } else if (type.toLowerCase().equals("void")) { 35 return BasicTypeInfo.VOID_TYPE_INFO; 36 } else if (type.toLowerCase().equals("biginteger")) { 37 return BasicTypeInfo.BIG_INT_TYPE_INFO; 38 } else if (type.toLowerCase().equals("bigdecimal")) { 39 return BasicTypeInfo.BIG_DEC_TYPE_INFO; 40 } else if (type.toLowerCase().equals("date")) { 41 return SqlTimeTypeInfo.DATE; 42 } else if (type.toLowerCase().equals("timestamp")) { 43 return SqlTimeTypeInfo.TIMESTAMP; 44 } else if (type.toLowerCase().equals("time")) { 45 return SqlTimeTypeInfo.TIME; 46 } else if (type.toLowerCase().equals("row")) { 47 //目前复杂的支持row,其它的待扩展 48 List<AthenaxJSONSchema> schemaList = schema.getSchema(); 49 List<String> nameList = new ArrayList<String>(); 50 List<TypeInformation<?>> typeInformationList = new ArrayList<TypeInformation<?>>(); 51 //有数据就继续处理 52 if (null != schemaList) { 53 for (AthenaxJSONSchema childSchema : schemaList) { 54 nameList.add(childSchema.getKey()); 55 typeInformationList.add(getTypeInfo(childSchema)); 56 } 57 } 58 //构造返回结果 59 String[] nameArray = new String[nameList.size()]; 60 nameList.toArray(nameArray); 61 TypeInformation<?>[] typeInformationArray = new TypeInformation<?>[typeInformationList 62 .size()]; 63 typeInformationList.toArray(typeInformationArray); 64 //return new RowTypeInfo(); 65 return Types.ROW(nameArray, typeInformationArray); 66 } else { 67 String errorMsg = "fail to find TypeInformation for type[" + type + "]"; 68 LoggerUtil.error(errorMsg); 69 throw new TableNotExistException(type, type); 70 } 71 } 72 73 // TableSchema SCHEMA = new TableSchema(new String[] { "id", "proctime" }, 74 // new TypeInformation<?>[] { BasicTypeInfo.INT_TYPE_INFO, 75 // SqlTimeTypeInfo.TIMESTAMP }); 76 public TableSchema getTableSchema() throws TableNotExistException { 77 String[] keys = new String[schemas.size()]; 78 TypeInformation<?>[] types = new TypeInformation<?>[schemas.size()]; 79 int index = 0; 80 for (AthenaxJSONSchema schema : schemas) { 81 keys[index] = schema.getKey(); 82 types[index++] = getTypeInfo(schema); 83 } 84 return new TableSchema(keys, types); 85 } 86 87 public String getTimeField() { 88 //目前只支持第1层定义,以后再优化 89 for (AthenaxJSONSchema schema : schemas) { 90 if (1 == schema.getTime()) { 91 return schema.getKey(); 92 } 93 } 94 return null; 95 } 96 97}
然后sql定义这么写
SELECT SUM(nested.number) as nestedNumber,hundredFunction(SUM(CAST(`value` AS DOUBLE))) as `sum`,COUNT(`value`) as `count`,AVG(CAST(`value` AS DOUBLE)) as `avg`,MAX(CAST(`value` AS DOUBLE)) as `max`,MIN(CAST(`value` AS DOUBLE)) as `min`,MAX(`time`) as `time` FROM input.tumble_topic_input_15 WHERE metric IS NOT NULL AND `value` IS NOT NULL and `time` IS NOT NULL GROUP BY metric,TUMBLE(`time`, INTERVAL '3' SECOND)
schema定义如下：
"schemas": [{"key": "metric","type": "string"},{"key": "time","type": "timestamp","time": 1},{"key":"value","type": "string"},{"key": "nested","type": "row","schema":[{"key": "clientId","type": "string"},{"key": "number","type": "int"}]}]
测试通过！