资源描述
单击此处编辑母版标题样式,单击此处编辑母版文本样式,第二级,第三级,第四级,第五级,2017-4-20,#,全国高校标准教材,云计算,姊妹篇，剖析大数据核心技术和实战应用,大数据,刘鹏主编张燕张重生张志立 副主编,BIG DATA,刘 鹏,教授，清华大学博士。现任南京大数据研究院院长、中国信息协会大数据分会副会长,、中国大数据技术与应用联盟副,理事长。,主持完成科研项目,25,项，发表论文,80,余篇，出版专业书籍,15,本。获部级科技进步二等奖,4,项、三等奖,4,项。主编的,云计算,被全国高校普遍采用，被引用量排名中国计算机图书第一名。创办了知名的中国云计算（,）和中国大数据（,）网站。,曾率队夺得,2002 PennySort,国际计算机排序比赛冠军，两次夺得全国高校科技比赛最高奖，并三次夺得清华大学科技比赛最高奖。,荣获“全军十大学习成才标兵”（排名第一）、南京“十大杰出青年”、江苏省中青年科学技术带头人、清华大学“学术新秀”等称号。,第二章,数据采集与预处理,2.1,2,.2,数据预处理原理,2.3,数据仓库与,ETL,工具,习题,全国高校标准教材,云计算,姊妹篇，剖析大数据核心技术和实战应用,大数据采集架构,of,42,3,2.1,大,数据采集架构,第二章 数据采集与预处理,如今，社会中各个机构、部门、公司、团体等正在实时不断地产生大量的信息，这些信息需要以简单的方式进行处理，同时又要十分准确且能迅速满足各种类型的数据（信息）需求者。这给我们带来了许多挑战，第一个挑战就是在大量的数据中收集需要的数据，下面介绍常用的大数据采集工具,。,of,42,4,2.1.1,概述,2.1,大,数据采集架构,第二章 数据采集与预处理,Flume,Chukwa,Scrible,Kafka,大,数据采集工具,of,42,5,2.1.2,常用大数据采集工具,数据采集最传统的方式是企业自己的生产系统产生的数据,，除,上述生产系统中的数据外，企业的信息系统还充斥着大量的用户行为数据、日志式的活动数据、事件信息等,，越来越,多的企业通过架设日志采集系统来保存这些数据，希望通过这些数据获取其商业或社会价值。,2.1,大,数据采集架构,第二章 数据采集与预处理,of,42,6,在,Flume,中，外部输入称为,Source,（源），系统输出称为,Sink,（接收端）。,Channel,（通道）把,Source,和,Sink,链接在,一起。,Apache,Chukwa,项目与,Flume,有些相类似,，,Chukwa,继承了,Hadoop,的伸缩性和鲁棒性,。也,内置一个功能强大的工具箱，用于显示系统监控和分析,结果,。,互联网时代，,网络爬虫,也是许多企业获取数据的一种方式。,Nutch,就是,网络爬虫中的娇娇者，,Nutch,是,Apache,旗下的开源项目，存在已经超过,10,年，拥有大量的忠实用户,。,Flume,体系架构,2.1,大,数据采集架构,第二章 数据采集与预处理,of,42,7,2.1.3 Apache,Kafka,数据采集,Apache Kafka,被设计成能够高效地处理大量实时数据，其特点是快速的、可扩展的、分布式的，分区的和可复制,的。,Kafka,是用,Scala,语言编写的，虽然置身于,Java,阵营，但其并不遵循,JMS,规范,。,Topics,（话题）：消息的分类名。,Producers,（消息发布者）：能够发布消息到,Topics,的进程。,Consumers,（消息接收者）：可以从,Topics,接收消息的进程。,Broker,（代理）：组成,Kafka,集群的单个节点。,基本,Kafka,集群的工作流程,2.1,大,数据采集架构,第二章 数据采集与预处理,of,42,8,1,、,Topics,Topics,是消息的分类名（或,Feed,的名称）。,Kafka,集群或,Broker,为每一个,Topic,都会维护一个分区日志。每一个分区日志是有序的消息序列，消息是连续追加到分区日志上，并且这些消息是不可更改的。,2,、日志区分,一个,Topic,可以有多个分区，这些分区可以作为并行处理的单元，从而使,Kafka,有能力高效地处理大量数据。,Topics,与日志分析,2.1,大,数据采集架构,第二章 数据采集与预处理,of,42,9,3,、,Producers,Producers,是向它们选择的主题发布数据。生产者可以选择分配某个主题到哪个分区上。这可以通过使用循环的方式或通过任何其他的语义分函数来实现。,4,、,Consumers,Kafka,提供一种单独的消费者抽象，此抽象具有两种模式的特征消费组：,Queuing,和,Publish-Subscribe,。,5,、,Apache Kafka,的安装及使用,因为,Kafka,是处理网络上请求，所以，应该为其创建一个专用的用户，这将便于对,Kafka,相关服务的管理，减少对服务器上其他服务的影响。,2.1,大,数据采集架构,第二章 数据采集与预处理,of,42,10,使用,useradd,命令来创建一个,Kafka,用户：,$,sudo useradd 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 update,Apache Kafka,需要,Java,运行环境，这里使用,apt-get,命令安装,default-jre,包，然后安装,Java,运行环境：,$sudo apt-get install default-jre,通过下面的命令测试一下,Java,运行环境是否安装成功，并查看,Java,的版本信息：,$java-version,2.1,大,数据采集架构,第二章 数据采集与预处理,of,42,11,机器有如下显示：,2.1,大,数据采集架构,第二章 数据采集与预处理,of,42,12,2.1,大,数据采集架构,第二章 数据采集与预处理,of,42,13,2.1,大,数据采集架构,第二章 数据采集与预处理,of,42,14,2.1,大,数据采集架构,第二章 数据采集与预处理,of,42,15,6,、,使用,Java,来编写,Kafka,的实例,首先，编写,KafkaProducer.properties,文件：,zk.connect =localhost:2181,broker.list =localhost:9092,serializer.class =kafka.serializer.StringEncoder,request.required.acks=1,下面的代码是使用,Java,编写了一个,Kafka,消息发布,者,：,import kafka.javaapi.producer.Producer;,import kafka.producer.KeyedMessage;,import kafka.producer.ProducerConfig;,public class MyKafkaProducer,private Producer producer;,private final String topic;,public MyKafkaProducer(String topic)throws Exception,InputStream in=Properties.class.,getResourceAsStream(KafkaProducer.properties);,Properties props=new Properties();,props.load(in);,ProducerConfig config=new ProducerConfig(props);,producer=new Producer(config);,public void sendMessage(String msg),KeyedMessage data=,new KeyedMessage(topic,msg);,producer.send(data);,producer.close();,public static void main(String args)throws Exception,MyKafkaProducer producer=new MyKafkaProducer(HelloTopic);,String msg=Hello Kafka!;,producer.sendMessage(msg);,2.1,大,数据采集架构,第二章 数据采集与预处理,of,42,16,下面创建,Comsumer,，首先编写,KafkaProperties,文件：,zk.connect =localhost:2181,group.id =testgroup,zookeeper.session.timeout.ms=500,zookeeper.sync.time.ms =250,mit.interval.ms =1000,上述参数配置，十分容易理解，具体的详细说明，可以参考,Kafka,的官方文档。下面的代码是使用,Java,编写了一个,Kafka,的,Comsumer,。,import java.io.InputStream;,import java.util.HashMap;,import java.util.List;,import 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 consumer;,private final String topic;,public MyKafkaConsumer(String topic)throws Exception,InputStream in=Properties.class.,getResourceAsStream(KafkaProducer.properties);,Properties props=new Properties();,props.load(in);,ConsumerConfig config=new ConsumerConfig(props);,consumer=Consumer.createJavaConsumerConnector(config);,this.topic=topic;,public void consumeMessage(),Map topicMap=new HashMap();,topicMap.put(topic,new Integer(1);,Map consumerStreamsMap=,consumer.createMessageStreams(topicMap);,List streamList=,consumerStreamsMap.get(topic);,for(final KafkaStream stream:streamLi
展开阅读全文