合作机构:阿里云 / 腾讯云 / 亚马逊云 / DreamHost / NameSilo / INWX / GODADDY / 百度统计
译者 | 李睿
审校 | 重楼
在当今竞争激烈的市场环境中,为了生存和发展,企业必须能够实时收集、处理和响应数据。无论是检测欺诈、个性化用户体验还是监控系统,现在都需要接近即时的数据。
然而,构建和运行任务关键型实时数据管道具有挑战性。基础设施必须具有容错性、无限可扩展性,并与各种数据源和应用程序集成。这就是ApacheKafka、Python和云平台的用武之地。
这个综合指南中将介绍:
这里将包括大量的代码片段、配置示例和文档链接,以便获得这些非常有用的技术的实践经验。
Apache Kafka是一个分布式、分区、复制的提交日志,用于可靠且大规模地存储数据流。Apache Kafka的核心是提供以下功能:
Kafka架构由以下主要组件组成:
消息被发布到名为“主题”的类别中。每个主题都充当消息提要或消息队列。常见的场景是每个消息类型或数据流的一个主题。Kafka主题中的每条消息都有一个唯一的标识符,称为偏移量,它代表了在主题中的位置。一个主题可以分为多个分区,这些分区是可以存储在不同代理上的主题片段。分区允许Kafka通过在多个消费者之间分配负载来扩展和并行化数据处理。
生产者是向Kafka主题发布消息的应用程序。它们连接到Kafka集群,序列化数据(例如JSON或Avro),分配一个密钥,并将其发送到适当的主题。
例如,一个Web应用程序可以产生点击流事件,或者一个移动应用程序可以产生使用统计。
消费者从Kafka主题中读取消息并进行处理。处理可能涉及解析数据、验证、聚合、过滤、存储到数据库等。
消费者连接到Kafka集群,并订阅一个或多个主题来获取消息提要,然后根据用例需求进行处理。
这是一个Kafka服务器,它接收来自生产者的消息,分配偏移量,将消息提交到存储中,并将数据提供给消费者。Kafka集群由多个代理组成,以实现可扩展性和容错性。
ZooKeeper处理代理之间的协调和共识,例如控制器选举和主题配置。它维护Kafka操作所需的集群状态和配置信息。
这涵盖了Kafka的基础知识。要深入了解,可以参考一些Kafka文档。
以下了解如何通过在云中运行Kafka来简化管理。
虽然Kafka具有高度可扩展性和可靠性,但它的运行涉及部署、基础设施管理、监控、安全、故障处理、升级等方面的大量工作。
值得庆幸的是,Kafka现在是所有主要云计算提供商提供的完全托管服务:
服务 | 描述 | 定价 |
AWS MSK | 在AWS上完全托管、高可用的Apache Kafka集群。处理基础设施,扩展,安全,故障处理等。 | 基于代理的数量 |
Google Cloud Pub/Sub | 基于Kafka的无服务器实时消息服务。自动扩展,至少一次交付保证。 | 基于使用指标 |
Confluent Cloud | 完全管理的事件流平台,由Apache Kafka提供支持。提供免费层。 | 基于功能的分层定价 |
Azure Event Hubs | Apache Kafka的高吞吐量事件摄取服务。与Azure数据服务的集成。 | 基于吞吐量单位 |
托管服务抽象了Kafka操作的复杂性,可以让用户专注数据管道。
接下来,将使用Python、Kafka和云平台构建一个实时管道。也可以参考以下的指南作为另一个示例。
Kafka的基本实时管道有两个主要组件:向Kafka发布消息的生产者和订阅主题并处理消息的消费者。
其架构遵循以下流程:
为了进行简化,将使用Confluent Kafka Python客户端库。
生产者应用程序从数据源收集数据并将其发布到Kafka主题。作为一个例子,假设有一个Python服务从一个Web应用程序收集用户点击流事件。
在Web应用程序中,当用户的行为像是页面浏览或产品评级时,可以捕获这些事件并将它们发送给Kafka。
可以抽象出Web应用程序如何收集数据的实现细节。
Python
from confluent_kafka import Producer
import json
# User event data
event = {
"timestamp": "2022-01-01T12:22:25",
"userid": "user123",
"page": "/product123",
"action": "view"
}
# Convert to JSON
event_json = json.dumps(event)
# Kafka producer configuration
conf = {
'bootstrap.servers': 'my_kafka_cluster-xyz.cloud.provider.com:9092',
'client.id': 'clickstream-producer'
}
# Create producer instance
producer = Producer(conf)
# Publish event
producer.produce(topic='clickstream', value=event_json)
# Flush and close producer
producer.flush()
producer.close()
TOP