json数据处理实战:Kafka+Flume+Morphline+Solr+Hue数据组合索引
#CDH的专有写法,开源版本不支持。zkHost : "$ZK_HOST"
}
morphlines : [
{
id : morphline1
importCommands : ["org.kitesdk.**", "org.apache.solr.**"]
commands : [
{
#Flume传过来的kafka的json数据是用二进制流的形式，需要先读取json
readJson{}
}
{
#读出来的json字段必须转换成filed才能被solr索引到
extractJsonPaths {
flatten:true
paths:{
account:/account
accountName:/accountName
subaccount:/subaccount
subaccountName:/subaccountName
eventTime:/timestamp
eventType:/eventType
eventTags:"/eventTags[]/name"
#按分钟存timestamp
eventTimeInMinute_tdt:/timestamp
#按小时存timestamp
eventTimeInHour_tdt:/timestamp
#按天存timestamp
eventTimeInDay_tdt:/timestamp
#_tdt后缀会被动态识别为日期类型的索引字段
#按不同时间间隔存索引以增加查询性能
}
}
}
#转换long型时间为Date格式
{convertTimestamp {
field : eventTimeInMinute_tdt
inputFormats : ["unixTimeInMillis"]
inputTimezone : UTC
outputFormat : "yyyy-MM-dd'T'HH:mm:ss.SSS'Z/MINUTE'"
outputTimezone : Asia/Shanghai
}}
{convertTimestamp {
field : eventTimeInHour_tdt
inputFormats : ["unixTimeInMillis"]
inputTimezone : UTC
outputFormat : "yyyy-MM-dd'T'HH:mm:ss.SSS'Z/HOUR'"
outputTimezone : Asia/Shanghai
}}
{convertTimestamp {
field : eventTimeInDay_tdt
inputFormats : ["unixTimeInMillis"]
inputTimezone : UTC
outputFormat : "yyyy-MM-dd'T'HH:mm:ss.SSS'Z/DAY'"
outputTimezone : Asia/Shanghai
}}
#kafka中的json数据传到flume中时会被放入_attachment_body字段，readJson后会变成JsonNode对象，需要toString之后才能保存
{toString { field : _attachment_body }}
#为每一条记录生成一个UUID
{generateUUID {
field :>
}}
#对未定义的Solr字段加tws前缀，根据schema.xml中定义的tws_*为text_ws类型，会动态未未定义的字段建索引。
{
sanitizeUnknownSolrFields {
# Location from which to fetch Solr schema
solrLocator : ${SOLR_LOCATOR}
renameToPrefix:"tws_"
}
}
#将数据导入到solr中
{loadSolr {solrLocator : ${SOLR_LOCATOR}}}
]
}
]
页:
[1]