资源预览内容
第1页 / 共41页
第2页 / 共41页
第3页 / 共41页
第4页 / 共41页
第5页 / 共41页
亲,该文档总共41页,到这儿已超出免费预览范围,如果喜欢就下载吧!
资源描述
,大数据,BIG DATA,全国高校标准教材云计算姊妹篇,剖析大数据核心技术和实战应用,大数据采集架构,of,42,2,2.1大数据采集架构,第二章 数据采集与预处理,如今,社会中各个机构、部门、公司、团体等正在实时不断地产生大量的信息,这些信息需要以简单的方式进行处理,同时又要十分准确且能迅速满足各种类型的数据(信息)需求者。这给我们带来了许多挑战,第一个挑战就是在大量的数据中收集需要的数据,下面介绍常用的大数据采集工具。,of,42,3,2.1.1概述,2.1大数据采集架构,第二章 数据采集与预处理,of,42,4,2.1.2 常用大数据采集工具,数据采集最传统的方式是企业自己的生产系统产生的数据,除上述生产系统中的数据外,企业的信息系统还充斥着大量的用户行为数据、日志式的活动数据、事件信息等,越来越多的企业通过架设日志采集系统来保存这些数据,希望通过这些数据获取其商业或社会价值。,2.1大数据采集架构,第二章 数据采集与预处理,of,42,5,在Flume中,外部输入称为Source(源),系统输出称为Sink(接收端)。Channel(通道)把Source和Sink链接在一起。 Apache Chukwa项目与Flume有些相类似,Chukwa继承了Hadoop的伸缩性和鲁棒性。也内置一个功能强大的工具箱,用于显示系统监控和分析结果。 互联网时代,网络爬虫也是许多企业获取数据的一种方式。Nutch就是网络爬虫中的娇娇者,Nutch是Apache旗下的开源项目,存在已经超过10年,拥有大量的忠实用户。,Flume体系架构,2.1大数据采集架构,第二章 数据采集与预处理,of,42,6,2.1.3 Apache Kafka数据采集,Apache Kafka被设计成能够高效地处理大量实时数据,其特点是快速的、可扩展的、分布式的,分区的和可复制的。Kafka是用Scala语言编写的,虽然置身于Java阵营,但其并不遵循JMS规范。,Topics(话题):消息的分类名。 Producers(消息发布者):能够发布消息到Topics的进程。 Consumers(消息接收者):可以从Topics接收消息的进程。 Broker(代理):组成Kafka集群的单个节点。,基本Kafka集群的工作流程,2.1大数据采集架构,第二章 数据采集与预处理,of,42,7,1、Topics,Topics是消息的分类名(或Feed的名称)。Kafka集群或Broker为每一个Topic都会维护一个分区日志。每一个分区日志是有序的消息序列,消息是连续追加到分区日志上,并且这些消息是不可更改的。,2、日志区分,一个Topic可以有多个分区,这些分区可以作为并行处理的单元,从而使Kafka有能力高效地处理大量数据。,Topics与日志分析,2.1大数据采集架构,第二章 数据采集与预处理,of,42,8,3、Producers,Producers是向它们选择的主题发布数据。生产者可以选择分配某个主题到哪个分区上。这可以通过使用循环的方式或通过任何其他的语义分函数来实现。,4、Consumers,Kafka提供一种单独的消费者抽象,此抽象具有两种模式的特征消费组:Queuing 和Publish-Subscribe。,5、Apache Kafka的安装及使用,因为Kafka是处理网络上请求,所以,应该为其创建一个专用的用户,这将便于对Kafka相关服务的管理,减少对服务器上其他服务的影响。,2.1大数据采集架构,第二章 数据采集与预处理,of,42,9,使用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,10,机器有如下显示:,2.1大数据采集架构,第二章 数据采集与预处理,of,42,11,2.1大数据采集架构,第二章 数据采集与预处理,of,42,12,2.1大数据采集架构,第二章 数据采集与预处理,of,42,13,2.1大数据采集架构,第二章 数据采集与预处理,of,42,14,6、使用Java来编写Kafka的实例,首先,编写KafkaProducer.properties文件:,zk.connect = localhost:2181 broker.list = localhost:9092 serializer.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,15,下面创建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 : 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 Exception String groupId = testgroup; String topic = HelloTopic; MyKafkaConsumer consumer = new MyKafkaConsumer(topic); consumer.consumeMessage(); ,全国高校标准教材云计算姊妹篇,剖析大数据核心技术和实战应用,大数据采集架构,of,42,16,2.2数据预处理原理,第二章 数据采集与预处理,通过数据预处理工作,可以使残缺的数据完整,并将错误的数据纠正、多余的数据去除,进而将所需的数据挑选出来,并且进行数据集成。数据预处理的常见方法有数据清洗、数据集成与数据变换。,of,42,17,2.2数据预处理原理,第二章 数据采集与预处理,数据属性分为数值属性和非数值属性进行处理,通过利用已存数据的多数信息来推测缺失值,数据属性分为数值属性和非数值属性进行处理,
网站客服QQ:2055934822
金锄头文库版权所有
经营许可证:蜀ICP备13022795号 | 川公网安备 51140202000112号