大數(shù)據(jù)采集與預(yù)處理

大數(shù)據(jù)采集與預(yù)處理第二章數(shù)據(jù)采集與預(yù)處理2.1

2.2數(shù)據(jù)預(yù)處理原理2.3數(shù)據(jù)倉庫與ETL工具習(xí)題大數(shù)據(jù)采集架構(gòu)of4222.1大數(shù)據(jù)采集架構(gòu)第二章數(shù)據(jù)采集與預(yù)處理如今，社會(huì)中各個(gè)機(jī)構(gòu)、部門、公司、團(tuán)體等正在實(shí)時(shí)不斷地產(chǎn)生大量的信息，這些信息需要以簡(jiǎn)單的方式進(jìn)行處理，同時(shí)又要十分準(zhǔn)確且能迅速滿足各種類型的數(shù)據(jù)（信息）需求者。這給我們帶來了許多挑戰(zhàn)，第一個(gè)挑戰(zhàn)就是在大量的數(shù)據(jù)中收集需要的數(shù)據(jù)，下面介紹常用的大數(shù)據(jù)采集工具。of4232.1.1概述2.1大數(shù)據(jù)采集架構(gòu)第二章數(shù)據(jù)采集與預(yù)處理FlumeChukwaScribleKafka大數(shù)據(jù)采集工具of4242.1.2常用大數(shù)據(jù)采集工具數(shù)據(jù)采集最傳統(tǒng)的方式是企業(yè)自己的生產(chǎn)系統(tǒng)產(chǎn)生的數(shù)據(jù)，除上述生產(chǎn)系統(tǒng)中的數(shù)據(jù)外，企業(yè)的信息系統(tǒng)還充斥著大量的用戶行為數(shù)據(jù)、日志式的活動(dòng)數(shù)據(jù)、事件信息等，越來越多的企業(yè)通過架設(shè)日志采集系統(tǒng)來保存這些數(shù)據(jù)，希望通過這些數(shù)據(jù)獲取其商業(yè)或社會(huì)價(jià)值。2.1大數(shù)據(jù)采集架構(gòu)第二章數(shù)據(jù)采集與預(yù)處理of425在Flume中，外部輸入稱為Source（源），系統(tǒng)輸出稱為Sink（接收端）。Channel（通道）把Source和Sink鏈接在一起。ApacheChukwa項(xiàng)目與Flume有些相類似，Chukwa繼承了Hadoop的伸縮性和魯棒性。也內(nèi)置一個(gè)功能強(qiáng)大的工具箱，用于顯示系統(tǒng)監(jiān)控和分析結(jié)果?；ヂ?lián)網(wǎng)時(shí)代，網(wǎng)絡(luò)爬蟲也是許多企業(yè)獲取數(shù)據(jù)的一種方式。Nutch就是網(wǎng)絡(luò)爬蟲中的嬌嬌者，Nutch是Apache旗下的開源項(xiàng)目，存在已經(jīng)超過10年，擁有大量的忠實(shí)用戶。Flume體系架構(gòu)2.1大數(shù)據(jù)采集架構(gòu)第二章數(shù)據(jù)采集與預(yù)處理of4262.1.3ApacheKafka數(shù)據(jù)采集ApacheKafka被設(shè)計(jì)成能夠高效地處理大量實(shí)時(shí)數(shù)據(jù)，其特點(diǎn)是快速的、可擴(kuò)展的、分布式的，分區(qū)的和可復(fù)制的。Kafka是用Scala語言編寫的，雖然置身于Java陣營，但其并不遵循JMS規(guī)范。Topics（話題）：消息的分類名。Producers（消息發(fā)布者）：能夠發(fā)布消息到Topics的進(jìn)程。Consumers（消息接收者）：可以從Topics接收消息的進(jìn)程。Broker（代理）：組成Kafka集群的單個(gè)節(jié)點(diǎn)?；綤afka集群的工作流程2.1大數(shù)據(jù)采集架構(gòu)第二章數(shù)據(jù)采集與預(yù)處理of4271、TopicsTopics是消息的分類名（或Feed的名稱）。Kafka集群或Broker為每一個(gè)Topic都會(huì)維護(hù)一個(gè)分區(qū)日志。每一個(gè)分區(qū)日志是有序的消息序列，消息是連續(xù)追加到分區(qū)日志上，并且這些消息是不可更改的。2、日志區(qū)分一個(gè)Topic可以有多個(gè)分區(qū)，這些分區(qū)可以作為并行處理的單元，從而使Kafka有能力高效地處理大量數(shù)據(jù)。Topics與日志分析2.1大數(shù)據(jù)采集架構(gòu)第二章數(shù)據(jù)采集與預(yù)處理of4283、ProducersProducers是向它們選擇的主題發(fā)布數(shù)據(jù)。生產(chǎn)者可以選擇分配某個(gè)主題到哪個(gè)分區(qū)上。這可以通過使用循環(huán)的方式或通過任何其他的語義分函數(shù)來實(shí)現(xiàn)。4、ConsumersKafka提供一種單獨(dú)的消費(fèi)者抽象，此抽象具有兩種模式的特征消費(fèi)組：Queuing和Publish-Subscribe。5、ApacheKafka的安裝及使用因?yàn)镵afka是處理網(wǎng)絡(luò)上請(qǐng)求，所以，應(yīng)該為其創(chuàng)建一個(gè)專用的用戶，這將便于對(duì)Kafka相關(guān)服務(wù)的管理，減少對(duì)服務(wù)器上其他服務(wù)的影響。2.1大數(shù)據(jù)采集架構(gòu)第二章數(shù)據(jù)采集與預(yù)處理of429使用useradd命令來創(chuàng)建一個(gè)Kafka用戶：

$sudouseraddkafka–m使用passwd命令來設(shè)置其密碼：$sudopasswdkafaka接下來把kafaka用戶添加到sudo管理組，以便kafaka用戶具有安裝ApacheKafka依賴庫的權(quán)限。這里使用adduser命令來進(jìn)行添加：$sudoadduserkafkasudo這時(shí)就可以使用kafka賬戶了。切換用戶可以使用su命令：$su-kafka在ApacheKafka安裝所依賴的軟件包前，最好更新一下apt管理程序的軟件列表：$sudoapt-getupdateApacheKafka需要Java運(yùn)行環(huán)境，這里使用apt-get命令安裝default-jre包，然后安裝Java運(yùn)行環(huán)境：$sudoapt-getinstalldefault-jre通過下面的命令測(cè)試一下Java運(yùn)行環(huán)境是否安裝成功，并查看Java的版本信息：$java-version2.1大數(shù)據(jù)采集架構(gòu)第二章數(shù)據(jù)采集與預(yù)處理of4210機(jī)器有如下顯示：2.1大數(shù)據(jù)采集架構(gòu)第二章數(shù)據(jù)采集與預(yù)處理of42112.1大數(shù)據(jù)采集架構(gòu)第二章數(shù)據(jù)采集與預(yù)處理of42122.1大數(shù)據(jù)采集架構(gòu)第二章數(shù)據(jù)采集與預(yù)處理of42132.1大數(shù)據(jù)采集架構(gòu)第二章數(shù)據(jù)采集與預(yù)處理of42146、使用Java來編寫Kafka的實(shí)例首先，編寫KafkaPperties文件：zk.connect=localhost:2181broker.list=localhost:9092serializer.class=kafka.serializer.StringEncoderrequest.required.acks=1下面的代碼是使用Java編寫了一個(gè)Kafka消息發(fā)布者：importducer.Producer;importducer.KeyedMessage;importducer.ProducerConfig;publicclassMyKafkaProducer{privateProducer<String,String>producer;privatefinalStringtopic;publicMyKafkaProducer(Stringtopic)throwsException{InputStreamin=Properties.class.getResourceAsStream("KafkaPperties");Propertiesprops=newProperties();props.load(in);ProducerConfigconfig=newProducerConfig(props);producer=newProducer<String,String>(config);}publicvoidsendMessage(Stringmsg){KeyedMessage<String,String>data=newKeyedMessage<String,String>(topic,msg);producer.send(data);producer.close();}publicstaticvoidmain(String[]args)throwsException{MyKafkaProducerproducer=newMyKafkaProducer("HelloTopic");Stringmsg="HelloKafka!";producer.sendMessage(msg);}}2.1大數(shù)據(jù)采集架構(gòu)第二章數(shù)據(jù)采集與預(yù)處理of4215下面創(chuàng)建Comsumer，首先編寫KafkaProperties文件：zk.connect=localhost:2181group.id=testgroupzookeeper.session.timeout.ms=500zookeeper.sync.time.ms=250erval.ms=1000上述參數(shù)配置，十分容易理解，具體的詳細(xì)說明，可以參考Kafka的官方文檔。下面的代碼是使用Java編寫了一個(gè)Kafka的Comsumer。importjava.io.InputStream;importjava.util.HashMap;importjava.util.List;importjava.util.Map;importjava.util.Properties;importkafka.consumer.ConsumerConfig;importkafka.consumer.ConsumerIterator;importkafka.consumer.KafkaStream;importkafka.javaapi.consumer.ConsumerConnector;importkafka.consumer.Consumer;

publicclassMyKafkaConsumer{privatefinalConsumerConnectorconsumer;privatefinalStringtopic;publicMyKafkaConsumer(Stringtopic)throwsException{InputStreamin=Properties.class.getResourceAsStream("KafkaPperties");Propertiesprops=newProperties();props.load(in);ConsumerConfigconfig=newConsumerConfig(props);consumer=Consumer.createJavaConsumerConnector(config);this.topic=topic;}publicvoidconsumeMessage(){Map<String,String>topicMap=newHashMap<String,String>();topicMap.put(topic,newInteger(1));Map<String,List<KafkaStream<byte[],byte[]>>>consumerStreamsMap=consumer.createMessageStreams(topicMap);List<KafkaStream<byte[],byte[]>>streamList=consumerStreamsMap.get(topic);for(finalKafkaStream<byte[],byte[]>stream:streamList){ConsumerIterator<byte[],byte[]>consumerIte=stream.iterator();while(consumerIte.hasNext())System.out.println("message::"+newString(consumerIte.next().message()));}if(consumer!=null)consumer.shutdown();}publicstaticvoidmain(String[]args)throwsException{StringgroupId="testgroup";Stringtopic="HelloTopic";MyKafkaConsumerconsumer=newMyKafkaConsumer(topic);consumer.consumeMessage();}}第二章數(shù)據(jù)采集與預(yù)處理2.1

2.2數(shù)據(jù)預(yù)處理原理2.3數(shù)據(jù)倉庫與ETL工具習(xí)題大數(shù)據(jù)采集架構(gòu)of42162.2數(shù)據(jù)預(yù)處理原理第二章數(shù)據(jù)采集與預(yù)處理通過數(shù)據(jù)預(yù)處理工作，可以使殘缺的數(shù)據(jù)完整，并將錯(cuò)誤的數(shù)據(jù)糾正、多余的數(shù)據(jù)去除，進(jìn)而將所需的數(shù)據(jù)挑選出來，并且進(jìn)行數(shù)據(jù)集成。數(shù)據(jù)預(yù)處理的常見方法有數(shù)據(jù)清洗、數(shù)據(jù)集成與數(shù)據(jù)變換。of42172.2數(shù)據(jù)預(yù)處理原理第二章數(shù)據(jù)采集與預(yù)處理填

充缺

失值數(shù)據(jù)屬性分為數(shù)值屬性和非數(shù)值屬性進(jìn)行處理，通過利用已存數(shù)據(jù)的多數(shù)信息來推測(cè)缺失值數(shù)據(jù)屬性分為數(shù)值屬性和非數(shù)值屬性進(jìn)行處理，通過利用已存數(shù)據(jù)的多數(shù)信息來推測(cè)缺失值大量采用同一屬性值，可能會(huì)誤導(dǎo)挖掘程序得出有偏差甚至錯(cuò)誤的結(jié)論數(shù)據(jù)偏離的問題小，但該方法十分費(fèi)時(shí),不具備實(shí)際的可操作性通常當(dāng)在缺少類標(biāo)號(hào)時(shí)，通過這樣的方法來填補(bǔ)缺失值利用均值替換缺失值忽略元組人工填寫缺失值使用一個(gè)全局常量填充缺失值用屬性的均值填充缺失值用同類樣本的屬性均值填充缺失值使用最可能的值填充缺失值of42182.2.1數(shù)據(jù)清洗2.2數(shù)據(jù)預(yù)處理原理第二章數(shù)據(jù)采集與預(yù)處理of4219分箱方法通過考察某一數(shù)據(jù)周圍數(shù)據(jù)的值，即“近鄰”來光滑有序數(shù)據(jù)的值。分箱01回歸02光滑數(shù)據(jù)可以通過一個(gè)函數(shù)擬合數(shù)據(jù)來實(shí)現(xiàn)。線性回歸的目標(biāo)就是查找擬合兩個(gè)屬性的“最佳”線，使得其中一個(gè)屬性可以用于預(yù)測(cè)出另一個(gè)屬性。聚類03離群點(diǎn)可通過聚類進(jìn)行檢測(cè)，將類似的值組織成群或簇，離群點(diǎn)即為落在簇集合之外的值。許多數(shù)據(jù)光滑的方法也是涉及離散化的數(shù)據(jù)歸約方法。噪聲是被測(cè)量的變量的隨機(jī)誤差或方差。給定一個(gè)數(shù)值屬性，如何才能使數(shù)據(jù)“光滑”，去掉噪聲？下面給出數(shù)據(jù)光滑技術(shù)的具體內(nèi)容。2.2數(shù)據(jù)預(yù)處理原理第二章數(shù)據(jù)采集與預(yù)處理of4220數(shù)據(jù)清洗可以視為一個(gè)過程，包括檢測(cè)偏差與糾正偏差兩個(gè)步驟：2糾正偏差1檢查偏差可以使用已有的關(guān)于數(shù)據(jù)性質(zhì)的知識(shí)發(fā)現(xiàn)噪聲、離群點(diǎn)和需要考察的不尋常的值。這種知識(shí)或“關(guān)于數(shù)據(jù)的數(shù)據(jù)”稱為元數(shù)據(jù)。即一旦發(fā)現(xiàn)偏差，通常需要定義并使用一系列的變換來糾正它們。但這些工具只支持有限的變換，因此，常?？赡苄枰獮閿?shù)據(jù)清洗過程的這一步編寫定制的程序。2.2數(shù)據(jù)預(yù)處理原理第二章數(shù)據(jù)采集與預(yù)處理（1）模式集成和對(duì)象匹配問題（2）冗余問題（3）元組重復(fù)（4）數(shù)據(jù)值沖突的檢測(cè)與處理問題數(shù)據(jù)挖掘經(jīng)常需要數(shù)據(jù)集成合并來自多個(gè)數(shù)據(jù)存儲(chǔ)的數(shù)據(jù)。數(shù)據(jù)還可能需要變換成適于挖掘的形式。數(shù)據(jù)分析任務(wù)多半涉及數(shù)據(jù)集成。問題of42212.2.2數(shù)據(jù)集成2.2數(shù)據(jù)預(yù)處理原理第二章數(shù)據(jù)采集與預(yù)處理1、光滑。去除數(shù)據(jù)中的噪聲2、聚集。對(duì)數(shù)據(jù)進(jìn)行匯總或聚集。3、

數(shù)據(jù)泛化。使用概念分層，用高層概念替換低層或“原始”數(shù)據(jù)4、規(guī)范化。將屬性數(shù)據(jù)按比例縮放，使之落入一個(gè)小的特定區(qū)間5、屬性構(gòu)造?？梢詷?gòu)造新的屬性并添加到屬性集中，以幫助挖掘過程of42222.2.3數(shù)據(jù)變換數(shù)據(jù)變換的目的是將數(shù)據(jù)變換或統(tǒng)一成適合挖掘的形式。數(shù)據(jù)變換主要涉及以下內(nèi)容：第二章數(shù)據(jù)采集與預(yù)處理2.1

2.2數(shù)據(jù)預(yù)處理原理2.3數(shù)據(jù)倉庫與ETL工具習(xí)題大數(shù)據(jù)采集架構(gòu)of42232.3數(shù)據(jù)倉庫與ETL工具第二章數(shù)據(jù)采集與預(yù)處理數(shù)據(jù)倉庫中的數(shù)據(jù)來自于多種業(yè)務(wù)數(shù)據(jù)源，這些數(shù)據(jù)源可能處于不同硬件平臺(tái)上，使用不同的操作系統(tǒng)，數(shù)據(jù)模型也相差很遠(yuǎn)。如何獲取并向數(shù)據(jù)倉庫加載這些數(shù)據(jù)量大、種類多的數(shù)據(jù)，已成為建立數(shù)據(jù)倉庫所面臨的一個(gè)關(guān)鍵問題。of42242.3.1數(shù)據(jù)倉庫與ETL工具數(shù)據(jù)倉庫，是在企業(yè)管理和決策中面向主題的、集成的、隨時(shí)間變化的、非易失性數(shù)據(jù)的集合。2.3數(shù)據(jù)倉庫與ETL工具第二章數(shù)據(jù)采集與預(yù)處理如何獲取并向數(shù)據(jù)倉庫加載數(shù)據(jù)量大、種類多的數(shù)據(jù)，一般要使用專業(yè)的數(shù)據(jù)抽取、轉(zhuǎn)換和裝載工具，這些工具合并起來被稱為ETL（Extract-Transform-Load）。InformaticaPowerCenter常用ETL工具IBMDatastageWarehouseBuilder（OWB）OracleDataIntegrator（ODI）MicrosoftSQLServerIntegrationServices開源Kettleof42252.3.2常用ETL工具2.3數(shù)據(jù)倉庫與ETL工具第二章數(shù)據(jù)采集與預(yù)處理of4226PowerCenterIBMDatastageKettleInformatica的PowerCenter是一個(gè)可擴(kuò)展、高性能企業(yè)數(shù)據(jù)集成平臺(tái)，應(yīng)用于各種數(shù)據(jù)集成流程，通過該平臺(tái)可實(shí)現(xiàn)自動(dòng)化、重復(fù)使用及靈活性IBMInfoSphereDataStage是一款功能強(qiáng)大的ETL工具，是IBM數(shù)據(jù)集成平臺(tái)IBMInformationServer的一部分，是專門的數(shù)據(jù)提取、數(shù)據(jù)轉(zhuǎn)換、數(shù)據(jù)發(fā)布的工具。Kettle是Pentaho中的ETL工具，Pentaho是一套開源BI解決方案。Kettle是一款國外優(yōu)秀的開源ETL工具，由純Java編寫，可以在Windows、Linux、UNIX上運(yùn)行，無須安裝，數(shù)據(jù)抽取高效穩(wěn)定。2.3數(shù)據(jù)倉庫與ETL工具第二章數(shù)據(jù)采集與預(yù)處理of422701（1）Chef可使用戶創(chuàng)