專6-4-基于kafka-spark streaming的數(shù)據(jù)處理系統(tǒng)及測試-甄麗霞

parkStreaming數(shù)據(jù)處理系統(tǒng)及測試待測系統(tǒng)涉及到的相關技術知234AnswerAnswerColume非頻繁更新項頻繁更新項5fastfastfastpartition01fastpartition02fastpartition01fastpartition02fastpartition03freqpartition01freqpartition02freqpartition03fastpartition05fastpartition08freqpartition05freqpartition06freqpartition07nfast-fast-partition09fastpartition10freqpartition08freqpartition10n6jsonFasthbaseFastjsonFasthbaseFast…map1map2map3kkkk7SSMSMS8zhihu-spooldir-kafka-agent.sources=zhihu-spooldir-source-freqzhihu-spooldir-source-freq3zhihu-spooldir-source-fastzhihu-spooldir-source-instantzhihu-spooldir-kafka-agent.channels=zhihu-kafka-channel-freqzhihu-kafka-channel-freq3zhihu-kafka-channel-fastzhihu-kafka-channel-instant1定義zhihu-freq、zhihu-freq3、zhihu-fast、zhihu-instant各組件對應的的配置項#zhihu-freqzhihu-spooldir-kafka-agent.sources.zhihu-spooldir-source-freq.inputCharset=UTF-8//注：編碼方式，默認是"UTF-8"zhihu-spooldir-kafka-agent.sources.zhihu-spooldir-source-freq.decodeErrorPolicy=IGNORE//注：傳輸?shù)倪^程中有不可解碼的流出現(xiàn)會導致flume停止服務，加上這個配置之后增加flume魯棒性zhihu-spooldir-kafka-agent.sources.zhihu-spooldir-source-freq.customSourceCounterType=TimedSourceCerceptors=timestamp-interceptorstatic-interceptorcirclenumber-interceptordocid-interceptor//注：在eventsheader中加入key和value的攔截器，包括時間戳、環(huán)號、erceptors.timestamp-interceptor.type=timestamp注：在header中加入時間戳erceptors.docid-interceptor.type=erceptor.zhihu.ZhihuDocIdInterceptor$Builder//注：調(diào)用docid算法，在header中加docidzhihu-spooldir-kafka-agent.sources.zhihu-spooldir-source-freq.channels=zhihu-kafka-channel-freq//注：確認本source對應的channel組件名稱zhihu-spooldir-kafka-agent.channels.zhihu-kafka-channel-freq.type=org.apache.flume.channel.kafka.KafkaChannel//注：定義channel的類型為Kafkachannel。zhihu-spooldir-kafka-agent.channels.zhihu-kafka-channel-freq.brokerList=10.1xx.1xx.29:xxxx//注：設定鏈接的kafka的ip和端口zhihu-spooldir-kafka-agent.channels.zhihu-kafka-channel-freq.topic=zhihu-freq//注：設定本source對應的kafka的topic類型是zhihu-freqzhihu-spooldir-kafka-agent.channels.zhihu-kafka-channel-freq.zookeeperConnect=10.1xx.1xx.29:xxxx/kafka//注：設定需要注冊和分配資源的zookeeper的ip和端口9zhihu-kafka-hdfszhihu-kafka-hdfs-agent.sinks.zhihu-hdfs-sink-freq.hdfs.rollInterval=120//注：hdfssink間隔多長將臨時文件滾動成最終目標文件，單位：秒；zhihu-kafka-hdfs-agent.sinks.zhihu-hdfs-sink-freq.hdfs.rollSize=0//注：當臨時文件達到該大小(單位：bytes)時，滾動成目標文件；如果設置成0，則表示不根據(jù)臨時文件大小來滾動文件；zhihu-kafka-hdfs-agent.sinks.zhihu-hdfs-sink-freq.hdfs.callTimeout=120000//注：執(zhí)行HDFS操作的超時時間(單位：毫秒)；zhihu-kafka-hdfs-agent.sinks.zhihu-hdfs-sink-freq.hdfs.batchSize=10000//注：每個批次刷新到HDFS上的events數(shù)量zhihu-kafka-hdfs-agent.sinks.zhihu-hdfs-sink-freq.channel=zhihu-kafka-channel-freq//注：該sink對應的channel的名稱-hdfs-sink-instantzhihu-hdfs-sink-blacklistChannels.brokerList=ip:portelfreq定義channel、sink組件，每個channel都對應一個sink配置：zhihu-kafka-hdfs-agent.channels=zhihu-kafka-channel-freq-localqueryzhihu-kafka-channel-freqzhihu-kafka-channel-freq3zhihu-kafka-channel-fastzhihu-kafka-channel-instantzhihu-kafka-channel-blacklistzhihu-kafka-hdfs-agent.channels.zhihu-kafka-channel-freq.topic=zhihu-freq//注：kafka里對應的topic類型zhihu-kafka-hdfs-agent.channels.zhihu-kafka-channel-freq.zookeeperConnect=10.1xx.1xx.29:xxxx/kafka//注：zookeeper的端口zhihu-kafka-hdfs-agent.channels.zhihu-kafka-channel-freq.groupId=zhihu-freq//注：consumer需要設置其所歸屬的groupidzhihu-kafka-hdfs-agent.channels.zhihu-kafka-channel-freq.kafka.fetch.message.max.bytes=2000000000//注：表示消息的最大大小，單位是字節(jié)zhihu-kafka-hdfs-agent.sinks.zhihu-hdfs-sink-freq.type=hdfs//注：sink組件取出channel隊列中的數(shù)據(jù)，存入相應類型的存儲文件系統(tǒng)。這里定義的是存儲系統(tǒng)的類型zhihu-kafka-hdfs-agent.sinks.zhihu-hdfs-sink-freq.hdfs.path=hdfs://sss/xxx/xxx/data/zhihu-test/%{pushtype}/sjs_100_29/%Y%m/%Y%m%d//注：寫入hdfs的路徑，包含文件系統(tǒng)標識“sss/xxx/xxx/data/”是hdfs的系統(tǒng)路徑；“zhihu-test”是為本次測試創(chuàng)建的hdfs的數(shù)據(jù)存儲目錄；“%{pushtype}”是在spooldir-kafka的配置文件(erceptors.static-interceptor.value)中設置的pushtype的值；“sjs_100_29”是本測試機的標識；“%Y%m/%Y%m%d”年月的目錄地址；//注：存儲到hdfs里的文件名稱：/sss/xxx/xxx/data/zhihu-test/freq/sjs_100_2/i26s0.1t/0p16e0=12/dfrefqs-2016012423-7.1453808288400.lzo(T3)(T3)(Producer)(push)(push)(push)(T3)(T3)(Producer)(push)(push)(push)(Broker)Real-time(T2)Hadoop(T1)Other(T1)ata(T3)(pull)(pull)(pull)(pull)(push)((8 (注：即是哪個原始文件的數(shù)據(jù))、circlenumber(注：根據(jù)數(shù)據(jù)類型，偽造各類型數(shù)據(jù)的url，計算環(huán)數(shù)，數(shù)據(jù)總共分8環(huán))、pushtype(注：該數(shù)據(jù)類型是freqinstantfast寫入kafkaT1-P1-R0-LT1-P2-R0T1-P1-R2T1-P2-R1-LT2-PT1-P1-R0-LT1-P2-R0T1-P1-R2T1-P2-R1-LT2-P1-R2T2-P2-R1T1-P1-R1T1-P2-R2T2-P1-R1-LT2-P2-R0-LT2-P1-R1T2-P1-R0-LT2-P2-R2T1-P3-RT1-P3-R0-L(new)加載PA解析的jar包；加載配置文件確認下游模塊；加載配置文件確認topic類型；instantSparkstreaming加載PA解析的jar包；加載配置文件確認下游模塊；加載配置文件確認topic類型；instantSparkstreamingfreq加載配置文件確認topic類型；加載hadoop數(shù)據(jù)；Sparkstreaming根根據(jù)kafka的key，獲取對應的hbase數(shù)據(jù)；解析數(shù)據(jù)拼xpage,發(fā)送給index模塊；將instant數(shù)據(jù)寫入hbase；寫入寫入habasememory處理的進程數(shù)memory處理的進程數(shù)d1--masteryarn-client\//注：以client方式連接到YARN集群，集群的定位由環(huán)境變量HADOOP_CONF_DIR定義，該方式1--driver-memory1G\//注：drivermemory并不是master分配了多少內(nèi)存，而是管理多少內(nèi)存。換言之就是為當前應用分配了多少內(nèi)存運行的集群隊列1--num-executors5\注：在yarn集群上啟動5個進程進行數(shù)據(jù)處理，其中一個進程讀取數(shù)據(jù)，剩余進程進行數(shù)據(jù)處理一定程度就會爆棧。一般先設置DM(drivermemory)，隨后根據(jù)集群情況、任務大小等實際情況來設置EM(executor一定程度就會爆棧。一般先設置DM(drivermemory)，隨后根據(jù)集群情況、任務大小等實際情況來設置EM(executorinstant序列化之后發(fā)送給剩余的進程進行數(shù)據(jù)處理，但flume對于序列化的操even以通過yarn執(zhí)行的時候涉及到數(shù)據(jù)分發(fā)到多個進程了，此時flumeeven函數(shù)進行修改；但本機運行的時候不涉及數(shù)據(jù)分發(fā)所以能確性測試：instantPAxpagexslt解析正確性；不同業(yè)務流程涉及不fre