Hadoop生态架构之kafka

1、定位：分布式的消息队列系统，同时提供数据分布式缓存功能（默认7天）

2、消息持久化到磁盘，达到O(1)访问速度，预读和后写，对磁盘的顺序访问（比内存访问还要快）

3、Storm（分布式的实时计算框架）

Kafka目标成为队列平台

4、基本组件：

Broker：每一台机器是一个Broker

Producer：日志消息生产者，主要写数据

Consumer：日志消息消费者，主要读数据

Topic：是虚拟概念，不同的consumer去指定的topic去读数据，不同producer可以往不同的topic去写

Partition：是实际概念，文件夹，是在Topic的基础上做了进一步分层

5、Partition功能：负载均衡，需要保证消息的顺序性

顺序性的保证：订阅消息是从头往后读取的，写消息是尾部追加，所以整体消息是顺序的

如果有多个partiton存在，可能会出现顺序不一致的情况，原因：每个Partition相互独立

6、Topic：逻辑概念

一个或多个Partition组成一个Topic

7、Partition以文件夹的形式存在

8、Partition有两部分组成：

（1）index log：（存储索引信息，快速定位segment文件）

（2）message log：（真实数据的所在）

9、HDFS多副本的方式来完成数据高可用

如果设置一个Topic，假设这个Topic有5个Partition，3个replication

Kafka分配replication的算法：

假设：

将第i个Partition分配到(i % N)个Broker上

将第i个Partition的第j个replication分配到( （i+j） % N)个Broker上

虽然Partition里面有多个replication

如果里面有M个replication，其中有一个是Leader，其他M-1个follower

10、zookeeper包系统的可用性，zk中会保存一些meta信息（topic）

11、物理上，不同的topic的消息肯定是分开存储的

12、偏移量——offset：用来定位数据读取的位置

13、kafka内部最基本的消息单位——message

14、传输最大消息message的size不能超过1M，可以通过配置来修改

15、Consumer Group

16、传输效率：zero-copy

0拷贝：减少Kernel和User模式上下文的切换

直接把disk上的data传输给socket，而不是通过应用程序来传输

17、Kafka的消息是无状态的，消费者必须自己维护已消费的状态信息（offset）

减轻Kafka的实现难度

18、Kafka内部有一个时间策略：SLA——消息保留策略（消息超过一定时间后，会自动删除）

19、交付保证：

at least once：至少一次（会有重复、但不丢失）

at most once：最多发送一次（不重复、但可能丢失）

exactly once：只有一次（最理想），目前不支持，只能靠客户端维护

20、Kafka集群里面，topic内部由多个partition（包含多个replication），达到高可用的目的：

日志副本：保证可靠性

角色：主、从

ISR：是一个集合，只有在集合中的follower，才有机会被选为leader

如何让leader知道follower是否成功接收数据（心跳，ack）

如果心跳正常，代表节点活着

21、怎么算“活着”

（1）心跳

（2）如果follower能够紧随leader的更新，不至于被落的太远

如果一旦挂掉，从ISR集合把该节点删除掉

前提：需要把zookeeper提前启动好

一、单机版

1、启动进程：

]# /bin/kafka-server-startsh config/serverproperties

2、查看topic列表：

]# /bin/kafka-topicssh --list --zookeeper localhost:2181

3、创建topic：

]# /bin/kafka-topicssh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic newyear_test

4、查看topic描述：

]# /bin/kafka-topicssh --describe --zookeeper localhost:2181 --topic newyear_test

5、producer发送数据：

]# /bin/kafka-console-producersh --broker-list localhost:9092 --topic newyear_test

6、consumer接收数据：

]# /bin/kafka-console-consumersh --zookeeper localhost:2181 --topic newyear_test --from-beginning

7、删除topic：

]# /bin/kafka-topicssh --delete --zookeeper localhost:2181 --topic newyear_test

二、集群版

在slave1和slave2上的brokerid一定设置不同

分别在slave1和slave2上开启进程：

/bin/kafka-server-startsh config/serverproperties

创建topic：

]# /bin/kafka-topicssh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 5 --topic newyear_many_test

1、实现一个consumer group

首先在不同的终端分别开启consumer，保证groupid一致

]# python consumer_kafkapy

执行一次producer：

]# python producer_kafkapy

2、指定partition发送数据

]# python producer_kafka_2py

3、指定partition读出数据

]# python consumer_kafka_2py

consumer_kafkapy：

producer_kafkapy：

consumer_kafka_2py：

producer_kafka_2py：

1新建/conf/kafka_test/flume_kafkaconf

2启动flume：

]# flume-ng agent -c conf -f /conf/kafka_test/flume_kafkaconf -n a1 -Dflumerootlogger=INFO,console

启动成功，如下图：

3测试：

11flume监控产生的数据：

]# for i in seq 1 100 ; do echo '====> '$i >> 1log ; done

12kafka消费数据：

]# /bin/kafka-console-consumersh --zookeeper localhost:2181 --topic topic_1013 --from-beginning

消费结果如下图：

主流的大数据分析平台构架

1 Hadoop

Hadoop 采用 Map Reduce 分布式计算框架，根据 GFS开发了 HDFS 分布式文件系统，根据 Big Table 开发了 HBase数据存储系统。Hadoop 的开源特性使其成为分布式计算系统的事实上的国际标准。Yahoo，Facebook，Amazon 以及国内的百度，阿里巴巴等众多互联网公司都以 Hadoop 为基础搭建自己的分布。

2 Spark

Spark 是在 Hadoop 的基础上进行了一些架构上的改良。Spark 与Hadoop 最大的不同点在于，Hadoop 使用硬盘来存储数据，而Spark 使用内存来存储数据，因此 Spark 可以提供超过 Hadoop 100 倍的运算速度。由于内存断电后会丢失数据，Spark不能用于处理需要长期保存的数据。

3 Storm

Storm是 Twitter 主推的分布式计算系统。它在Hadoop的基础上提供了实时运算的特性，可以实时的处理大数据流。不同于Hadoop和Spark，Storm不进行数据的收集和存储工作，它直接通过网络实时的接受数据并且实时的处理数据，然后直接通过网络实时的传回结果。

　4Samza

Samza 是由 Linked In 开源的一项技术，是一个分布式流处理框架，专用于实时数据的处理，非常像Twitter的流处理系统Storm。不同的是Samza 基于 Hadoop，而且使用了 Linked In 自家的 Kafka 分布式消息系统。

Samza 非常适用于实时流数据处理的业务，如数据跟踪、日志服务、实时服务等应用，它能够帮助开发者进行高速消息处理,同时还具有良好的容错能力。

在当今大数据的应用越来越广泛的情况下，数据治理一直是企业面临的巨大问题。

大部分公司只是单纯的对数据进行了处理，而数据的血缘，分类等等却很难实现，市场上也急需要一个专注于数据治理的技术框架，这时Atlas应运而生。

Atlas官网地址： >

1、Hadoop

Hadoop 采用 Map Reduce 分布式计算框架，根据 GFS开发了 HDFS 分布式文件系统，根据 Big Table 开发了 HBase数据存储系统。Hadoop 的开源特性使其成为分布式计算系统的事实上的国际标准。Yahoo，Facebook，Amazon 以及国内的百度，阿里巴巴等众多互联网公司都以 Hadoop 为基础搭建自己的分布。

2、Spark

Spark 是在 Hadoop 的基础上进行了一些架构上的改良。Spark 与Hadoop 最大的不同点在于，Hadoop 使用硬盘来存储数据，而Spark 使用内存来存储数据，因此 Spark 可以提供超过 Hadoop 100 倍的运算速度。由于内存断电后会丢失数据，Spark不能用于处理需要长期保存的数据。

3、 Storm

Storm 是 Twitter 主推的分布式计算系统。它在Hadoop的基础上提供了实时运算的特性，可以实时的处理大数据流。不同于Hadoop和Spark，Storm不进行数据的收集和存储工作，它直接通过网络实时的接受数据并且实时的处理数据，然后直接通过网络实时的传回结果。

4、Samza

Samza 是由 Linked In 开源的一项技术，是一个分布式流处理框架，专用于实时数据的处理，非常像Twitter的流处理系统Storm。不同的是Samza 基于 Hadoop，而且使用了 Linked In 自家的 Kafka 分布式消息系统。

Samza 非常适用于实时流数据处理的业务，如数据跟踪、日志服务、实时服务等应用，它能够帮助开发者进行高速消息处理,同时还具有良好的容错能力。

Hadoop分布式文件系统（HDFS）是一种运行在通用硬件上的分布式文件系统。它与传统的分布式文件系统有很多相似之处，但是也有显著的不同。HDFS是高容错的，可以部署在低成本硬件上。HDFS提供了对应用数据的高吞吐量访问，适用于具有大数据集的应用。HDFS为了流数据访问放松了一些POSIX的限制。

HDFS是主从结构。一个HDFS集群由一个NameNode和一组DataNode组成。NameNode是主服务器，负责管理文件系统命名空间以及客户端对文件的访问。DataNode通常每个节点一个，负责管理存储。HDFS对外暴露了一个文件系统命名空间并允许用户数据作为文件存储。在内部实现上，一个文件会被分割成一个或多个block，这些block存储在一组DataNode上。NameNode负责执行文件系统命名空间 *** 作，例如打开，关闭，重命名文件和目录等。此外NameNode还维护着block和DataNode之间的映射关系。DataNode负责处理来自客户端的读写请求，并根据NameNode的指令创建，删除，备份block。

NameNode和DataNode都是运行在通用机器上的软件。这些机器通常使用Linux系统。HDFS使用Java构建，任何支持Java的机器都可以运行NameNode和DataNode。一种典型的集群部署方式是使用一台机器运行NameNode，其它机器每台运行一个DataNode实例。

HDFS使用传统的分层文件结构。用户可以创建目录并在目录下存储文件。文件系统命名空间结构与传统文件系统类似，用户可以创建，删除文件，将文件从一个目录移动到另一个目录，重命名文件。HDFS支持用户限额和访问权限。

NameNode维护整个文件系统命名空间，它会记录任何对命名空间的修改。应用程序可以指定HDFS中文件的备份数量。文件的拷贝数称为该文件的备份因子。这个信息也存储在NameNode中。

HDFS可以跨机器存储海量文件。每个文件分成一个block的序列存储。为了容错，文件的block会被备份。每个文件的block大小和备份因子都是可配置的。

文件中所有block的大小是相等的（除了最后一个），而对append和hsync提供可变长block支持后，用户可以直接创建一个新block，不必继续填充最后一个block。

应用程序可以指定文件的备份数。备份因子可在文件创建时指定，也可以稍后修改。HDFS的文件都是一次写入的（除了append和truncate），并且任何时候都只有一个写入器。

NameNode决定如何备份block。它周期性的接收来自DataNode的心跳检测和block报表。收到心跳检测说明DataNode工作正常，block报表包含该DataNode上的所有block。

备份文件的位置对HDFS的可用性和性能至关重要。对备份的优化让HDFS从众多分布式系统中脱颖而出。这个工作需要大量的优化和经验。机架感知备份放置策略的目的是提高数据的可靠性，可用性和网络带宽利用率。目前的备份放置策略实现是这个方向上的第一步。短期目标是在生产环境上对其进行验证，更多的了解它的行为，为测试和研究更复杂的策略奠定基础。

大型HDFS集群的机器通常隶属于多个机架。两个不同机架上的节点进行通信必须通过交换机。一般来说，同一机架机器之间的网络带宽要优于不同机架机器间的网络带宽。

NameNode通过Hadoop Rack Awareness进程确定每个DataNode所属的机架ID。一个简单但是并非最优的策略是将备份放置在独立的机架上。这种策略可以避免机架故障时丢失数据，读数据时也可以利用多个机架的网络带宽。这种策略在集群中平均分配备份文件，这样组件发生故障时可以平衡负载。但是这种策略会增加写入成本，因为数据需要跨机架传输。

最常见的情况，备份因子是3。HDFS的放置策略是：如果写入器位于DataNode上，则将副本放置在本地计算机，否则随机选择一个DataNode，另一个副本放置在另一个远程机架的节点上，最后一个副本放在同一个远程机架的另一个节点上。这种策略减少了机架间的写入流量，从而提高写性能。机架发生故障的几率远小于节点故障几率。这种策略并不影响数据可靠性和可用性，但是它确实减少了读 *** 作时的聚合网络带宽，因为一个block被放置到两个机架上而不是三个。这种策略的文件副本并不是均匀的分布在所有机架上，副本的三分之一位于一个节点，剩下的三分之二位于另一个机架上。这种策略可以提高写性能，而不会影响数据可靠性和读性能。

如果备份因子大于3，那么第四个和之后的副本随机放置，同时要保证副本数量不能超过机架的上限（公式： (replicas - 1) / racks + 2 ）。

由于DataNode不能放置同一个block的多个副本，所以最大备份因子就是最大DataNode数。

在提供了存储类型和存储策略的支持之后，除了机架感知，NameNode放置副本时也会考虑放置策略。NameNode首先根据机架感知选择节点，然后根据备份文件的放置策略检查该节点的存储类型，如果该候选节点没有要求的存储类型，NameNode会查找下一个节点。如果第一轮没有找到足够的节点放置备份，NameNode会使用后备存储类型开始第二轮查找。

目前，副本放置策略依然在开发中。

为了减少带宽消耗和读延迟，HDFS会尝试找寻一个离读请求最近的副本。如果读请求节点所在机架有这样一个副本，HDFS就优先使用这个副本。如果HDFS集群跨越多个数据中心，则本地数据中心的副本优先于远程副本。

启动HDFS时，NameNode会进入一种称为安全模式的特殊状态。安全模式下数据block无法备份。NameNode会从DataNode接收心跳检测和block报表。block报表包含该DataNode下所有数据block的列表信息。每个block都有一个指定的最小备份数。只有block的最小备份数登记到NameNode中后，block才可以备份。备份登记结束后，NameNode退出安全模式。这是如果还有block不满足最小备份数的条件，NameNode才开始备份这些block。

HDFS命名空间由NameNode保存，NameNode使用一个称为EditLog的事务日志记录对文件系统元数据的所有更改。例如，创建一个新文件会在EditLog中插入一条对应记录，同样的，修改文件备份因子也会插入一条记录。NameNode使用本地文件存储EditLog。整个文件系统命名空间，包括文件与block之间的映射关系，文件系统数据等，都保存在FsImage文件中。

NameNode在内存中维护文件系统命名空间和文件block映射关系的镜像。当NameNode启动，或者某个阈值触发了检查点时，NameNode从磁盘上读取FsImage和EditLog的内容，将所有EditLog中的事务 *** 作应用到FsImage的内存镜像中，然后在磁盘上生成一个全新的FsImage。之后可以截断EditLog，因为所有事务都已持久化到FsImage。这个过程称为检查点。检查点的目的是通过获取文件系统元数据的快照并保存到FsImage来保证HDFS文件系统元数据的一致性。读取FsImage可能很快，但是持续编辑FsImage就不同了。因此我们将 *** 作记录到EditLog中，而不是直接修改FsImage。在检查点期间，所有EditLog *** 作应用到FsImage。检查点可以按周期触发（ dfsnamenodecheckpointperiod ），也可以按事务数触发（ dfsnamenodecheckpointtxns ）。如果两个属性都设置了，第一个满足的阈值会触发检查点。

DataNode在本地文件系统中存储HDFS数据。DataNode对HDFS文件一无所知，它以block为单位存储HDFS数据。DataNode不会在同一个目录下保存所有文件。相反，它使用启发式方法来确定每个目录的最佳文件数，并适时创建子目录。在同一个目录下创建所有文件并不是最佳选择，因为本地文件系统可能无法支持一个目录下的大量文件。DataNode启动时，它会扫描整个本地文件系统，生成一个本地文件与数据block之间的关系列表，将其发送给NameNode，这个列表称为block报告。

所有HDFS通信协议都构建在TCP/IP协议之上。客户端通过TCP端口与NameNode建立连接，它使用ClientProtocol与NameNode交互。DataNode使用DataProtocol与NameNode交互。一个RPC抽象封装了客户端协议和DataNode协议。NameNode从不初始化任何RPC，它只是响应来自的客户端和DataNode的请求。

HDFS的主要目标是即使出现故障也可以可靠的存储数据。三种常见的故障分别是：NameNode故障，DataNode故障和网络分区。

DataNode周期性的发送心跳检测给NameNode。网络分区可能导致某些DataNode无法连接NameNode。NameNode无法收到DataNode的心跳检测后，它会把这样的DataNode标记为dead，并不在发送新的I/O请求。注册到死亡DataNode上的数据对HDFS来说不再可用，也会导致某些block的备份数少于文件指定的最小备份数。NameNode持续追踪block的备份情况并在必要时初始化备份 *** 作。重备份的原因是多种多样的：DataNode不可用，某个备份文件损坏，DataNode磁盘故障，或者文件的备份因子增大。

为了避免DataNode状态抖动引起的备份风暴，标记DataNode死亡的超时时间设置的很长（默认超过10分钟）。用户可以设置一个更短的时间将DataNode标记为陈旧（stale），这样可以避免对性能敏感的工作负载的陈旧DataNode的读写 *** 作。

HDFS架构与数据重平衡scheme兼容。scheme可以在DataNode的磁盘空间低于某个阈值时将数据移动到另一个DataNode上。如果对某个文件的需求特别高，scheme还可以动态创建额外的副本并平衡到整个集群中。这些数据平衡scheme还未实现。

从DataNode中读取的block可能是损坏的。损坏的原因有多种：磁盘故障，网络故障，或者软件问题。HDFS客户端会对文件内容进行校验和检查。当客户端创建一个HDFS文件时，它会计算出文件所有block的校验和并保存在同一个命名空间的一个独立的隐藏文件中。当客户单检索文件时还要检查对应校验和文件中的值。如果校验和不匹配，客户端会尝试该block其它节点上的副本。

FsImage和EditLog是HDFS的核心数据结构。如果它们发生损坏，HDFS就无法使用了。因此，可以通过配置让NameNode维护多个FsImage和EditLog的拷贝。对两个文件的修改会同步到所有拷贝中。这种同步 *** 作会降低NameNode的TPS，但是这种牺牲是可接受的，因为HDFS是数据密集，不是元数据密集。NameNode重启时，它会选择最一致的FsImage和EditLog使用。

另一种减低故障的办法是使用HA。

（略）

HDFS的目的是支持大型文件。HDFS支持一次写入多次读取。一个典型的block大小是128MB。因此，HDFS文件按照128MB的大小分割，每个block可能分布在不同的节点上。

客户端向HDFS文件写入数据时，如果备份因子是三，NameNode使用备份目标选择算法检索出一组DataNode。这个列表是可以存储副本的DataNode。客户端先向第一个DataNode写入数据，DataNode接收数据并将数据传输到列表中的第二个DataNode。第二个DataNode开始接收数据并继续传输数据到第三个DataNode。这样，数据通过管道从一个DataNode传输到下一个。

（略）

如果开启了trash配置，从FS shell中删除的文件并不会立刻从HDFS中删除，HDFS将它移动到一个trash目录（每个用户都有自己的trash目录， /user/<username>/Trash ）。只要文件还在trash目录中就可以快速恢复。

最近删除的文件移动到 /user/<username>/Trash/Current 目录中，每隔一段时间，HDFS会为这些文件创建检查点文件（ /user/<username>/Trash/<date> ）并删除旧检查点文件。

如果trash中的文件过期了，NameNode将这些文件从命名空间中删除。与文件关联的block被释放。删除文件和空间释放之间可能会有延迟。

下面是一个例子，首先创建两个文件：

然后删除test1，该文件会被移到Trash目录：

接着跳过Trash删除test2：

现在可以查看Trash目录：

文件的备份因子降低后，NameNode选择可以删除的副本，在下次心跳检测时把信息发送给DataNode，之后DataNode删除block并释放空间。

1、Hadoop是一个由Apache基金会所开发的分布式系统基础架构。用户可以在不了解分布式底层细节的情况下，开发分布式程序。充分利用集群的威力进行高速运算和存储。Hadoop实现了一个分布式文件系统（Hadoop Distributed File System），简称HDFS。HDFS有高容错性的特点，并且设计用来部署在低廉的（low-cost）硬件上；而且它提供高吞吐量（high throughput）来访问应用程序的数据，适合那些有着超大数据集（large data set）的应用程序。

2、HDFS放宽了（relax）POSIX的要求，可以以流的形式访问（streaming access）文件系统中的数据。Hadoop的框架最核心的设计就是：HDFS和MapReduce。HDFS为海量的数据提供了存储，而MapReduce则为海量的数据提供了计算。

hadoop是做分布式系统基础架构。

Hadoop是一个由Apache基金会所开发的分布式系统基础架构，一个能够对大量数据进行分布式处理的软件框架； Hadoop以一种可靠、高效、可伸缩的方式进行数据处理；用户可以在不了解分布式底层细节的情况下，开发分布式程序。

用户可以轻松地在Hadoop上开发和运行处理海量数据的应用程序。

Hadoop实现了一个分布式文件系统（Hadoop Distributed File System），简称HDFS。HDFS有高容错性的特点，并且设计用来部署在低廉的（low-cost）硬件上；而且它提供高吞吐量（high throughput）来访问应用程序的数据，适合那些有着超大数据集（large data set）的应用程序。HDFS放宽了（relax）POSIX的要求，可以以流的形式访问（streaming access）文件系统中的数据。

Hadoop的框架最核心的设计就是：HDFS和MapReduce。HDFS为海量的数据提供了存储，而MapReduce则为海量的数据提供了计算。

Hadoop主要有以下优点：

高可靠性。Hadoop按位存储和处理数据的能力值得人们信赖。

高扩展性。Hadoop是在可用的计算机集簇间分配数据并完成计算任务的，这些集簇可以方便地扩展到数以千计的节点中。

高效性。Hadoop能够在节点之间动态地移动数据，并保证各个节点的动态平衡，因此处理速度非常快。高容错性。Hadoop能够自动保存数据的多个副本，并且能够自动将失败的任务重新分配。

低成本。与一体机、商用数据仓库以及QlikView、Yonghong Z-Suite等数据集市相比，hadoop是开源的，项目的软件成本因此会大大降低。

Hadoop带有用Java语言编写的框架，因此运行在 Linux 生产平台上是非常理想的。Hadoop 上的应用程序也可以使用其他语言编写，比如 C++。

以上就是关于Hadoop生态架构之kafka全部的内容，包括:Hadoop生态架构之kafka、大数据分析的框架有哪些，各自有什么特点、Atlas介绍等相关内容解答，如果想了解更多相关内容，可以关注我们，你们的支持是我们更新的动力！