1、數據采集和預處理方式高效培訓教案 教授，清華大學博士。現任南京大數據研究院院長、中國信息協會大數據分會副會長、中國大數據技術與應用聯盟副理事長。 主持完成科研項目25項，發表論文80余篇，出版專業書籍15本。獲部級科技進步二等獎4項、三等獎4項。主編的被全國高校普遍采用，被引用量排名中國計算機圖書第一名。創辦了知名的中國云計算（）和中國大數據（）網站。 曾率隊奪得2002 PennySort國際計算機排序比賽冠軍，兩次奪得全國高校科技比賽最高獎，并三次奪得清華大學科技比賽最高獎。 榮獲“全軍十大學習成才標兵”（排名第一）、南京“十大杰出青年”、江蘇省中青年科學技術帶頭人、清華大學“學術新秀”等

2、稱號。第二章數據采集與預處理2.12.2數據預處理原理2.3數據倉庫與ETL工具習題全國高校標準教材?云計算?姊妹篇，剖析大數據核心技術和實戰應用大數據采集架構of4232.1大數據采集架構第二章 數據采集與預處理如今，社會中各個機構、部門、公司、團體等正在實時不斷地產生大量的信息，這些信息需要以簡單的方式進行處理，同時又要十分準確且能迅速滿足各種類型的數據信息需求者。這給我們帶來了許多挑戰，第一個挑戰就是在大量的數據中收集需要的數據，下面介紹常用的大數據采集工具。of424概述2.1大數據采集架構第二章 數據采集與預處理FlumeChukwaScribleKafka大數據采集工具of4252

3、.1.2 常用大數據采集工具數據采集最傳統的方式是企業自己的生產系統產生的數據，除上述生產系統中的數據外，企業的信息系統還充滿著大量的用戶行為數據、日志式的活動數據、事件信息等，越來越多的企業通過架設日志采集系統來保存這些數據，希望通過這些數據獲取其商業或社會價值。2.1大數據采集架構第二章 數據采集與預處理of426在Flume中，外部輸入稱為Source源，系統輸出稱為Sink接收端。Channel通道把Source和Sink鏈接在一起。Apache Chukwa工程與Flume有些相類似，Chukwa繼承了Hadoop的伸縮性和魯棒性。也內置一個功能強大的工具箱，用于顯示系統監控和分析結

4、果。互聯網時代，網絡爬蟲也是許多企業獲取數據的一種方式。Nutch就是網絡爬蟲中的嬌嬌者，Nutch是Apache旗下的開源工程，存在已經超過10年，擁有大量的忠實用戶。Flume體系架構2.1大數據采集架構第二章 數據采集與預處理of4272.1.3 Apache Kafka數據采集Apache Kafka被設計成能夠高效地處理大量實時數據，其特點是快速的、可擴展的、分布式的，分區的和可復制的。Kafka是用Scala語言編寫的，雖然置身于Java陣營，但其并不遵循JMS標準。Topics話題：消息的分類名。Producers消息發布者：能夠發布消息到Topics的進程。Consumers消

5、息接收者：可以從Topics接收消息的進程。Broker代理：組成Kafka集群的單個節點。根本Kafka集群的工作流程2.1大數據采集架構第二章 數據采集與預處理of428 1、TopicsTopics是消息的分類名或Feed的名稱。Kafka集群或Broker為每一個Topic都會維護一個分區日志。每一個分區日志是有序的消息序列，消息是連續追加到分區日志上，并且這些消息是不可更改的。 2、日志區分一個Topic可以有多個分區，這些分區可以作為并行處理的單元，從而使Kafka有能力高效地處理大量數據。Topics與日志分析2.1大數據采集架構第二章 數據采集與預處理of429 3、Produ

6、cersProducers是向它們選擇的主題發布數據。生產者可以選擇分配某個主題到哪個分區上。這可以通過使用循環的方式或通過任何其他的語義分函數來實現。 4、ConsumersKafka提供一種單獨的消費者抽象，此抽象具有兩種模式的特征消費組：Queuing 和Publish-Subscribe。 5、Apache Kafka的安裝及使用因為Kafka是處理網絡上請求，所以，應該為其創立一個專用的用戶，這將便于對Kafka相關效勞的管理，減少對效勞器上其他效勞的影響。2.1大數據采集架構第二章 數據采集與預處理of4210使用useradd命令來創立一個Kafka用戶：$sudo userad

7、d kafka m使用passwd 命令來設置其密碼：$sudo passwd kafaka接下來把kafaka用戶添加到sudo管理組，以便kafaka用戶具有安裝Apache Kafka依賴庫的權限。這里使用adduser命令來進行添加：$sudo adduser kafka sudo這時就可以使用kafka賬戶了。 切換用戶可以使用su命令：$su - kafka在Apache Kafka安裝所依賴的軟件包前，最好更新一下apt管理程序的軟件列表：$sudo apt-get updateApache Kafka需要Java運行環境，這里使用apt-get命令安裝default-jre包，

8、然后安裝Java運行環境：$sudo apt-get install default-jre通過下面的命令測試一下Java運行環境是否安裝成功，并查看Java的版本信息：$java -version2.1大數據采集架構第二章 數據采集與預處理of4211機器有如下顯示：2.1大數據采集架構第二章 數據采集與預處理of42122.1大數據采集架構第二章 數據采集與預處理of42132.1大數據采集架構第二章 數據采集與預處理of42142.1大數據采集架構第二章 數據采集與預處理of42156、使用Java來編寫Kafka的實例首先，編寫KafkaPperties文件：zk

9、.connect = localhost:2181broker.list = localhost:9092request.required.acks = 1下面的代碼是使用Java編寫了一個Kafka消息發布者：import ducer.Producer;import ducer.KeyedMessage;import ducer.ProducerConfig;public class MyKafkaProducer private Producer producer;private final String topic;p

10、ublic MyKafkaProducer(String topic) throws Exception InputStream in = Properties.class.getResourceAsStream(KafkaPperties);Properties props = new Properties();props.load(in);ProducerConfig config = new ProducerConfig(props);producer = new Producer(config);public void sendMessage(String msg

11、)KeyedMessage data = new KeyedMessage( topic, msg);producer.send(data);producer.close();public static void main(String args) throws ExceptionMyKafkaProducer producer = new MyKafkaProducer(HelloTopic);String msg = Hello Kafka!;producer. sendMessage(msg);2.1大數據采集架構第二章 數據采集與預處理of4216下面創立Comsumer，首先編寫Ka

12、fkaProperties文件：zk.connect = localhost:2181group.id = testgroupzookeeper.session.timeout.ms = 500zookeeper.sync.time.ms = 250auto erval.ms = 1000上述參數配置，十分容易理解，具體的詳細說明，可以參考Kafka的官方文檔。下面的代碼是使用Java編寫了一個Kafka的Comsumer。import java.io.InputStream;import java.util.HashMap;import java.util.List;impor

13、t java.util.Map;import java.util.Properties;import kafka.consumer.ConsumerConfig;import kafka.consumer.ConsumerIterator;import kafka.consumer.KafkaStream;import kafka.javaapi.consumer.ConsumerConnector;import kafka.consumer.Consumer;public class MyKafkaConsumer private final ConsumerConnector consum

14、er;private final String topic;public MyKafkaConsumer(String topic) throws ExceptionInputStream in = Properties.class.getResourceAsStream(KafkaPperties);Properties props = new Properties();props.load(in);ConsumerConfig config = new ConsumerConfig(props);consumer = Consumer.createJavaConsum

15、erConnector(config);this.topic = topic;public void consumeMessage() Map topicMap = new HashMap();topicMap.put(topic, new Integer(1);MapString, ListKafkaStream consumerStreamsMap =consumer.createMessageStreams(topicMap);ListKafkaStream streamList = consumerStreamsMap.get(topic);for (final KafkaStream

16、 stream : streamList) ConsumerIterator consumerIte = stream.iterator();while (consumerIte.hasNext()System.out.println(message : + new String(consumerIte.next().message();if (consumer != null)consumer.shutdown();public static void main(String args) throws ExceptionString groupId = testgroup;String to

17、pic = HelloTopic;MyKafkaConsumer consumer = new MyKafkaConsumer(topic);consumer.consumeMessage();第二章數據采集與預處理2.12.2數據預處理原理2.3數據倉庫與ETL工具習題全國高校標準教材?云計算?姊妹篇，剖析大數據核心技術和實戰應用大數據采集架構of42172.2數據預處理原理第二章 數據采集與預處理通過數據預處理工作，可以使殘缺的數據完整，并將錯誤的數據糾正、多余的數據去除，進而將所需的數據挑選出來，并且進行數據集成。數據預處理的常見方法有數據清洗、數據集成與數據變換。of42182.2數據

18、預處理原理第二章 數據采集與預處理填 充缺 失值數據屬性分為數值屬性和非數值屬性進行處理，通過利用已存數據的多數信息來推測缺失值數據屬性分為數值屬性和非數值屬性進行處理，通過利用已存數據的多數信息來推測缺失值大量采用同一屬性值，可能會誤導挖掘程序得出有偏差甚至錯誤的結論數據偏離的問題小，但該方法十分費時,不具備實際的可操作性通常當在缺少類標號時，通過這樣的方法來填補缺失值利用均值替換缺失值忽略元組人工填寫缺失值使用一個全局常量填充缺失值用屬性的均值填充缺失值用同類樣本的屬性均值填充缺失值使用最可能的值填充缺失值of42192.2.1 數據清洗2.2數據預處理原理第二章 數據采集與預處理of42

19、20分箱方法通過考察某一數據周圍數據的值，即“近鄰”來光滑有序數據的值。分箱01回歸02光滑數據可以通過一個函數擬合數據來實現。線性回歸的目標就是查找擬合兩個屬性的“最佳”線，使得其中一個屬性可以用于預測出另一個屬性。聚類03離群點可通過聚類進行檢測，將類似的值組織成群或簇，離群點即為落在簇集合之外的值。許多數據光滑的方法也是涉及離散化的數據歸約方法。噪聲是被測量的變量的隨機誤差或方差。給定一個數值屬性，如何才能使數據“光滑，去掉噪聲？下面給出數據光滑技術的具體內容。2.2數據預處理原理第二章 數據采集與預處理of4221數據清洗可以視為一個過程，包括檢測偏差與糾正偏差兩個步驟：2糾正偏差1檢

20、查偏差可以使用已有的關于數據性質的知識發現噪聲、離群點和需要考察的不尋常的值。這種知識或“關于數據的數據稱為元數據。即一旦發現偏差，通常需要定義并使用一系列的變換來糾正它們。但這些工具只支持有限的變換，因此，常常可能需要為數據清洗過程的這一步編寫定制的程序。2.2數據預處理原理第二章 數據采集與預處理1模式集成和對象匹配問題2冗余問題3元組重復4數據值沖突的檢測與處理問題數據挖掘經常需要數據集成合并來自多個數據存儲的數據。數據還可能需要變換成適于挖掘的形式。數據分析任務多半涉及數據集成。問題of42222.2.2 數據集成2.2數據預處理原理第二章 數據采集與預處理 1、光滑。去除數據中的噪聲

21、 2、聚集。對數據進行匯總或聚集。3、 數據泛化。使用概念分層，用高層概念替換低層或“原始數據 4、標準化。將屬性數據按比例縮放，使之落入一個小的特定區間5、屬性構造。可以構造新的屬性并添加到屬性集中，以幫助挖掘過程of42232.2.3 數據變換數據變換的目的是將數據變換或統一成適合挖掘的形式。數據變換主要涉及以下內容：第二章數據采集與預處理2.12.2數據預處理原理2.3數據倉庫與ETL工具習題全國高校標準教材?云計算?姊妹篇，剖析大數據核心技術和實戰應用大數據采集架構of42242.3 數據倉庫與ETL工具第二章 數據采集與預處理數據倉庫中的數據來自于多種業務數據源，這些數據源可能處于不

22、同硬件平臺上，使用不同的操作系統，數據模型也相差很遠。如何獲取并向數據倉庫加載這些數據量大、種類多的數據，已成為建立數據倉庫所面臨的一個關鍵問題。of42252.3.1 數據倉庫與ETL工具數據倉庫，是在企業管理和決策中面向主題的、集成的、隨時間變化的、非易失性數據的集合。2.3 數據倉庫與ETL工具第二章 數據采集與預處理如何獲取并向數據倉庫加載數據量大、種類多的數據，一般要使用專業的數據抽取、轉換和裝載工具，這些工具合并起來被稱為ETLExtract-Transform-Load。Informatica PowerCenter常用ETL工具IBM DatastageWarehouse Bu

23、ilder（OWB）Oracle Data Integrator（ODI）Microsoft SQLServer Integration Services開源Kettleof42262.3.2 常用ETL工具2.3 數據倉庫與ETL工具第二章 數據采集與預處理of4227PowerCenterIBM DatastageKettleInformatica的PowerCenter是一個可擴展、高性能企業數據集成平臺，應用于各種數據集成流程，通過該平臺可實現自動化、重復使用及靈活性IBM InfoSphere DataStage是一款功能強大的ETL工具，是IBM數據集成平臺IBM Informat

24、ion Server的一局部，是專門的數據提取、數據轉換、數據發布的工具。Kettle是Pentaho中的ETL工具，Pentaho是一套開源BI解決方案。Kettle是一款國外優秀的開源ETL工具，由純Java編寫，可以在Windows、Linux、UNIX上運行，無須安裝，數據抽取高效穩定。2.3 數據倉庫與ETL工具第二章 數據采集與預處理of422801（1）Chef可使用戶創建任務（Job）。它是提供圖形用戶界面的工作設計工具。02（2）Kitchen可使用戶批量使用由Chef設計的任務，一般在自動調度時借助此命令調用調試成功的任務。它是一個后臺運行的程序，以命令行方式，沒有圖形用戶界面。03（3）Spoon可使用戶通過圖形界面來設計ETL轉換過程，一般在編寫和調試ETL時用到。04（4）Span可使用戶批量運行由Spoon設計的ETL轉換，Span是一個后臺執行的程序，以命令行方式，沒有圖形界面，一般在自動調度時借助此命令調用調試成功的轉換。Kettle目前包括如下4個產品：2.3 數據倉庫與ETL工具第二章 數據采集與預處理案例：Kettle數據遷移可以在Kettle的官網下載進入主界面在“輸入文件夾下選擇“表輸入，并把它拖動到右側編輯區of42292.3.3 案例：Kettle數據遷移2.