《数据采集技术》课件-Kafka环境搭建

项目5

动态网页访问日志数据采集使用Flume采集某官网访问日志数据目录Content1使用Kafka消费Flume中某官网访问日志数据

2项目导言零

日志中通常包含了大量的用户访问网站的行为信息，比如页面响应时间、消耗的流量、访问时间、停留时间、是否访问成功等等一系列的信息，网站运营者可通过分析这些信息，对网站运营做出一定的决策。学习目标零知识目标了解什么是日志采集；熟悉日志数据采集的方法；掌握Flume进行数据采集的方法；精通使用Kafka进行消息订阅发布的方法技能目标具备使用Flume拦截器过滤数据的能力；具备熟悉Kafka的脚本操作的能力；具备掌握Flume采集日志文件的能力；具备掌握使用Kafka进行消息发布订阅的能力；素养目标具备精通Flume+Kafka架构实现数据采集的能力。技能目标具备事业心和责任心；具有艰苦奋斗的精神和务实作风；具有健康的体魄，良好的身体素质；具备责任心。使用Kafka消费Flume中某官网访问日志数据贰1Kafka简介Kafka是由Apache软件基金会开发的一个开源流处理平台。是一个快速、可扩展的、分布式的、分区的、可复制的基于zookeeper协调的分布式日志系统，用于web/nginx日志、访问日志、消息服务等，Linkedin于2010年将Kafka贡献给了Apache基金会并成为顶级开源项目。持久性、可靠性高并发高吞吐量、低延迟容错性可扩展性23451使用Kafka消费Flume中某官网访问日志数据贰1Kafka简介使用Kafka消费Flume中某官网访问日志数据贰2Kafka配置1将zookeeper安装包上传到“/usr/local”目录下并解压重命名为zookeeper2将模板文件“zoo_sample.cfg”文件复制重命名为“zoo.cfg”3进入zookeeper安装目录的“bin”目录，启动zookeeper并查看进程5在Kafka的“/bin”目录下后台启动Kafka进程4将Kafka安装包上传到“/usr/local”目录下解压并重命名为kafka使用Kafka消费Flume中某官网访问日志数据贰3Kafka脚本操作Kafka为开发人员提供了众多脚本对Kafka进行管理和相关的操作，其中常用的脚本包含Topic脚本、生产者脚本、消费者脚本等。使用Kafka消费Flume中某官网访问日志数据贰3Kafka脚本操作1.Topic脚本Topic脚本文件名为“kafka-topic.sh”，通过该脚本文件能够完成创建主题、查看主题、修改主题、删除主题等操作。参数描述--alter修改主题--config<String:name=value>创建或修改主题时，用于设置主题级别的参数--create创建主题--delete删除主题--describe查看主题的详细信息--disable-rack-aware创建主题时不考虑机架信息--help打印帮助信息--if-exists修改或删除主题时使用，只有当主题存在时才会执行动作--if-not-exists创建主题时使用，只有主题不存在时才会执行动作--list列出所有可用的主题--partitons分区数创建主题或增加分区时指定的分区数--replica-assignment分配方案手工指定分区副本分配方案--replication-factor副本数创建主题时指定副本数--topic主题名称指定主题名称--topics-with-overrides使用describe查看主题信息时，只展示包含覆盖配置的主题--unavailable-partitions使用describe查看主题信息时，只展示包含没有leader副本的分区--under-replicated-partitions使用describe查看主题信息时，只展示包含失效副本的分区--bootstrap-server链接的服务器地址和端口（必需）使用Kafka消费Flume中某官网访问日志数据贰3Kafka脚本操作2.生产者脚本

生产者脚本文件名为“kafka-console-producer.sh”，该脚本能够生产创建消息并将消息发送到消费者。参数描述--bootstrap-server要连接的服务器必需(除非指定–broker-list)--topic(必需)topic名称--batch-size单个批处理中发送的消息数（默认值：16384）--compression-codec压缩编解码器，可选值“none”、“gzip”、“snappy”、“lz4”或“zstd”。如果未指定值，则默认为“gzip”--max-block-ms在发送请求期间，生产者将阻止的最长时间（默认值：60000）--max-memory-bytes生产者用来缓冲等待发送到服务器的总内存(默认值:33554432)--max-partition-memory-bytes为分区分配的缓冲区大小(默认值:16384)--message-send-max-retries最大的重试发送次数(默认值:3)--metadata-expiry-ms强制更新元数据的时间阈值(ms)(默认值:300000)--producer-property将自定义属性传递给生成器的机制--producer.config生产者配置属性文件[--producer-property]优先于此配置配置文件完整路径--property自定义消息读取器--request-required-acks生产者请求的确认方式--request-timeout-ms生产者请求的确认超时时间--retry-backoff-ms生产者重试前，刷新元数据的等待时间阈值--socket-buffer-sizeTCP接收缓冲大小--timeout消息排队异步等待处理的时间阈值--sync同步发送消息使用Kafka消费Flume中某官网访问日志数据贰3Kafka脚本操作3.消费者脚本

消费者脚本文件名为“kafka-console-consumer.sh”，该脚本接收并消费生产者发送的消息。参数描述--group指定消费者所属组的ID--topic被消费的topic--partition指定分区；除非指定–offset，否则从分区结束(latest)开始消费--offset执行消费的起始offset位置;默认值:latest;/latest/earliest/偏移量--whitelist正则表达式匹配topic；–topic就不用指定了;匹配到的所有topic都会消费;当然用了这个参数,--partition--offset等就不能使用了--consumer-property将用户定义的属性以key=value的形式传递给使用者--consumer.config消费者配置属性文件请注意，[consumer-property]优先于此配置--property初始化消息格式化程序的属性--from-beginning从存在的最早消息开始，而不是从最新消息开始,注意如果配置了客户端名称并且之前消费过，那就不会从头消费了--max-messages消费的最大数据量，若不指定，则持续消费下去--skip-message-on-error如果处理消息时出错，请跳过它而不是暂停--isolation-level设置为read_committed以过滤掉未提交的事务性消息,设置为read_uncommitted以读取所有消息,默认值:read_uncommitted使用Kafka消费Flume中某官网访问日志数据贰4KafkaPythonAPI生产者API顾名思义，就是指应用程序发送数据流到Kafka集群中的topic中，供消费者消费使用，生产者是线程安全的，跨线程共享单个生产者实例通常比拥有多个实例更快。1.ProducerAPI（生产者API）kafka.KafkaProducer（关键字参数）参数描述bootstrap_server要连接的服务器必需(除非指定–broker-list)batch-size单个批处理中发送的消息数（默认值：16384）compression_type生产者生成的所有数据的压缩类型。有效值为“gzip”、“snappy”、“lz4”或“无”max_block_ms在发送请求期间，生产者将阻止的最长时间（默认值：60000）buffer_memory生产者用来缓冲等待发送到服务器的总内存(默认值:33554432)metadata_max_age_ms强制更新元数据的时间阈值(ms)(默认值:300000)--request-timeout-ms生产者请求的确认超时时间,默认值：30000--retry-backoff-ms生产者重试前，刷新元数据的等待时间阈值,默认值：100使用Kafka消费Flume中某官网访问日志数据贰4KafkaPythonAPI消费者API主要用于接收生产者集群发出的消息，并且消费者不是线程安全的，不应使用跨线程共享。2.ConsumerAPI（消费者API）kafka.KafkaConsumer(topics,

关键字参数)参数描述topics(str)要订阅的可选主题列表bootstrap_servers要连接的服务器client_id(str)此客户端的名称。此字符串在每个请求中传递给服务器，可用于标识与此客户端对应的特定服务器端日志条目group_id消费者组ID使用Kafka消费Flume中某官网访问日志数据贰4KafkaPythonAPI通过消费者API成功接收生产者消息后，会返回一个消息集合，每条消息会包含topic、消息所在的分区、消息所在分区的位置、消息的内容、消息的key，对应获取方法如表所示。方法描述partition消息所在的分区offset消息所在分区的位置key消息的keyvalue消息的内容

在“/usr/local/inspur/code/pykafka”目录中编写python代码，使用API创建消费者接收“test-topic”中的消息。[root@masterpykafka]#vimconsumer-demo.py#代码如下consumer=KafkaConsumer('test-topic',bootstrap_servers='localhost:9092')formessageinconsumer:print("topic=%s,partition=%d,offset=%d,value=%s"%(message.topic,message.partition,message.offset,message.value))[root@masterpykafka]#pyth