AI 驱动的开发者（MEAP）（二）（1）https://developer.aliyun.com/article/1516328
5.2 使用 Kafka 实时监控我们的资产
我们将实时监控我们的资产，以激励我们探索将生成式人工智能与事件驱动架构结合使用的探索。我们应该认为，信息安全资产管理系统外部的某些系统在我们的资产从一个位置移动到另一个位置时会触发事件。
要深入了解 ITAM 事件，我们需要配置一些额外的服务。在这种情况下，我们将使用 Apache Kafka。Apache Kafka 是一个分布式流平台，用于构建实时数据管道和流应用程序。它被设计用于处理来自多个来源的数据流，并将它们传送到多个消费者，有效地充当我们实时数据的中间人。
首先，我们将询问 Copilot Chat 如何在本地使用 Docker 运行 Kafka。Apache Kafka 有一个不应该存在的声誉，即安装和配置很困难。在 Docker 中运行将允许我们回避这个争议。使用 Copilot Chat，我们可以生成一个 docker compose 文件。然而，通常情况下，版本非常旧，甚至不支持一些硬件。图示 5.13 是从 Confluent（提供 Kafka 商业支持的公司）官方 GitHub 存储库中提取的更新的列表。请注意，docker-compose 文件的内容包括 Kafka 和 Zookeeper。Zookeeper 是 Kafka 用来管理和协调集群内代理的分布式协调服务，至少目前是这样。未来版本的目标是消除对 Zookeeper 的依赖。
图示 5.13 Docker-Compose 文件，用于启动带有 Zookeeper 的 Kafka
version: '2.1' services: zookeeper: image: confluentinc/cp-zookeeper:7.3.2 container_name: zookeeper ports: - "2181:2181" environment: ZOOKEEPER_CLIENT_PORT: 2181 ZOOKEEPER_SERVER_ID: 1 ZOOKEEPER_SERVERS: zoo1:2888:3888 kafka: image: confluentinc/cp-kafka:7.3.2 hostname: kafka container_name: kafka ports: - "9092:9092" - "29092:29092" - "9999:9999" environment: KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka:19092,EXTERNAL://${DOCKER_HOST_IP:-127.0.0.1}:9092,DOCKER://host.docker.internal:29092 KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT,DOCKER:PLAINTEXT KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181" KAFKA_BROKER_ID: 1 KAFKA_LOG4J_LOGGERS: "kafka.controller=INFO,kafka.producer.async.DefaultEventHandler=INFO,state.change.logger=INFO" KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1 KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1 KAFKA_JMX_PORT: 9999 KAFKA_JMX_HOSTNAME: ${DOCKER_HOST_IP:-127.0.0.1} KAFKA_AUTHORIZER_CLASS_NAME: kafka.security.authorizer.AclAuthorizer KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: "true" depends_on: - zookeeper
现在我们的新 Kafka 实例正在运行，我们现在需要一个消费者，它将从主题asset_location（我们很快将创建的主题）中拉取更新的位置，并在数据库中更新库存。同样，我们可以请求 Copilot Chat 为我们提供如何做到这一点的建议。
图示 5.14 提示 Copilot Chat 如何使用 Python 订阅主题
In our ITAM project, we would like a Python consumer for the assets_location Kafka topic. Once it receives a new updated location, it should update the domain model to reflect the new location. How should we do this?
Copilot Chat 生成的代码创建一个消费者，监听主题，并使用对AssetManager类的引用来更新受影响资产的位置。
图示 5.15 Copilot Chat 代码，用于监控资产位置
from kafka import KafkaConsumer from asset_manager import AssetManager consumer = KafkaConsumer( 'asset_location', bootstrap_servers=['localhost:9092'], auto_offset_reset='earliest', enable_auto_commit=True, group_id='my-group', value_deserializer=lambda x: tuple(map(float, x.decode('utf-8').split(','))) ) asset_manager = AssetManager() for message in consumer: asset_id, latitude, longitude = message.value asset_manager.update_asset_location(asset_id, (latitude, longitude))
但是，适配器应该知道AssetManager吗？虽然AssetManager不直接位于领域模型内部，但它是一个至关重要的中介。适配器和AssetManager之间的这种耦合程度可能会违反六边形架构模型中的六边形。让我们让 Copilot Chat 来发表意见。我们可以问它：“考虑到这个项目使用了六边形架构，让适配器知道AssetManager是一个好主意吗？”
图示 5.2 GitHub Copilot Chat 承认它实际上并不完美
看起来 Copilot Chat 认为这确实会被视为适配器的适当承包责任的违约。我们可以向我们的AssetManager类添加一个新的端口，它可以使用 Kakfa 端口。然而，让我们看看 Copilot Chat 是否有其他建议：“我们不希望 Kafka 消费者直接与AssetManager进行交互。有什么方法可以实现这一点？” Copilot Chat 建议我们应用Mediator模式，这类似于一个端口和适配器的组合。
中介者模式
Mediator 设计模式是一种行为模式，通过将对象之间的交互封装在一个中介对象中，促进对象之间的松耦合。中介对象充当一个中心枢纽，促进对象之间的通信，而无需对象知道彼此。
让我们首先看一下 Copilot Chat 生成的 Mediator 类。该类将位于AssetManager和 Kafka 消费者之间。
图 5.16 Mediator 类，它将调和 AssetManager 和 Kafka 消费者之间的交互
class AssetLocationMediator: def __init__(self): self.handlers = {} def register_handler(self, event_type, handler): if event_type not in self.handlers: self.handlers[event_type] = [] self.handlers[event_type].append(handler) def publish(self, event): event_type = type(event) if event_type in self.handlers: for handler in self.handlers[event_type]: handler(event)
这个实现与消息队列或消息发布者十分相似，这正是关键所在：解耦。
注意
有慧眼的读者可能会注意到我们在资产和硬件类型的资产之间的区别上有一点疏忽。在原始领域模型中，只有硬件有位置。通常，人们不会认为软件有位置。当然，您可以说软件安装在它的位置上，但人们是否会认为这个论点说服力有待商榷。不过，随着项目的继续进行，领域模型变平的原因是为了简化，因为在持久性层中的多态结构是一个令人分心的复杂主题。
现在我们有了一个将 Kafka 消费者与 AssetManager 解耦的策略，我们应该更新 Kafka 消费者以利用它。我们需要在构造函数中将中介者传递给类。这样，AssetManager 和消费者将可以访问相同的实例，并且消息可以自由地来来往往；或者，在这种情况下，流程将是单向的。您应该注意，我们打算在这个主题上读取和写入 JSON，因此我们需要让我们的值反序列化器理解这一点。
图 5.17 将中介者整合到 Kafka 消费者类中
from kafka import KafkaConsumer from itam.domain.events.asset_location_updated import AssetLocationUpdated import json class AssetLocationKafkaConsumer: def __init__(self, mediator): self.mediator = mediator self.consumer = KafkaConsumer( 'asset_location', bootstrap_servers=['localhost:9092'], enable_auto_commit=True, group_id='itam-group', value_deserializer=lambda m: json.loads(m.decode('utf-8')) ) def poll(self): print("Polling for asset location updates") for message in self.consumer: asset_id = message.value['asset_id'] latitude = message.value['latitude'] longitude = message.value['longitude'] timestamp = message.value['timestamp'] event = AssetLocationUpdated(asset_id, latitude, longitude, timestamp) self.mediator.publish(event)
接下来，我们将检查AssetManager类需要的更改，以包含跟踪这些位置的功能。您应该注意，要使该项目完全运行起来，您需要修改AssetManager、SQLAlchemyAssetRepository和Asset类，并在数据库中创建一个名为itam.asset_locations的新表。完整且更新的源代码可在本书的 GitHub 存储库中找到。现在，我们将专注于为我们的系统流动的事件所需的更改，并在需要时使用存储库作为参考。
图 5.3 AssetManager需要另外一个构造函数参数和一个方法来处理其位置对象的更新
对于AssetManager类，有两个必需的更改：首先，我们需要将AssetLocationMediator添加到构造函数中，并注册它来处理AssetLocationUpdated事件。其次，我们需要添加一个处理此事件的方法。在这种情况下，我们称此方法为update_asset_location。删节后的代码如下。
图 5.18 更新后的AssetManager类的构造函数和事件处理程序
from itam.infrastructure.mediators.asset_location_mediator import class AssetManager: def __init__(self, base_repository: BaseRepository[Asset], mediator: AssetLocationMediator): self._repository = base_repository self.mediator = mediator self.mediator.register_handler(AssetLocationUpdated, self.update_asset_location) def update_asset_location(self, event: AssetLocationUpdated) -> None: asset = self.read(event.asset_id) asset.add_location(event.latitude, event.longitude, event.timestamp) #self.update(asset) print(f"Asset {asset.id} location updated to {event.latitude}, {event.longitude} at {event.timestamp}")
Asset类的add_location方法仅仅是将新的位置追加到位置列表的末尾。更复杂的领域模型可能会包括一个current_location属性，将其余部分归类为历史位置列表；然而，鉴于我们试图使我们的事件在系统中流动，使事情保持简单是明智的。
我们待办事项清单上只有一项最终项目：创建主题。如何创建主题呢？这是一个很好的问题。幸运的是，我们运行的 Docker 容器中有我们需要的所有工具。所以，让我们登录到我们的 Kafka Docker 实例中。我们使用以下命令（假设您的 Docker 实例命名为 kafka）：docker exec -it kafka /bin/bash.
应该首先检查的是是否已经创建了任何主题。我们可以使用以下命令来执行：kafka-topics --list --bootstrap-server localhost:9092. 此命令将列出在此 Kafka 集群上运行的所有现有主题。正如您所见，没有任何主题。
需要一个主题，让我们来创建它。您可以使用以下命令：kafka-topics --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic asset_location. 如果您重新运行kafka-topics --list命令，您将看到新主题。我们在创建主题命令中包含的分区和复制因子说明告诉 Kafka 我们希望有一个分区和一个复制因子。如果我们是为生产环境或测试以外的任何目的设置这个，我们可能希望它们大于那个数，以确保数据的可用性。
表 5.1 Kafka 控制台命令摘要
| 动作 | 命令 |
| 创建 |
kafka-topics --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic asset_location
|
| 读取 |
kafka-console-consumer --broker-list localhost:9092 --topic asset_location –from-beginning
|
| 写入 |
kafka-console-producer --broker-list localhost:9092 --topic asset_location
|
| 删除 |
kafka-topics --delete --topic asset_location --bootstrap-server localhost:9092
|
| 列出 |
kafka-topics --list --bootstrap-server localhost:9092
|
现在是有趣的部分，观察应用程序的运行情况。Kafka 自带一个控制台生产者，允许我们从标准输入向 Kafka 发布消息。要做到这一点，我们应该使用以下命令启动控制台生产者：kafka-console-producer --broker-list localhost:9092 --topic asset_location。您将进入一个交互式会话，可以在每行发布一条消息。让我们发布一些消息，模拟我们的资产在芝加哥附近移动的情况。
列表 5.19 Kafka 控制台生产者的条目
{"asset_id": 1, "latitude": 41.8781, "longitude": -87.6298, "timestamp": "2022-01-01T00:00:00Z"} {"asset_id": 1, "latitude": 41.9000, "longitude": -87.6244, "timestamp": "2022-01-01T00:10:00Z"} {"asset_id": 1, "latitude": 41.8676, "longitude": -87.6270, "timestamp": "2022-01-01T00:20:00Z"} {"asset_id": 1, "latitude": 41.8788, "longitude": -87.6359, "timestamp": "2022-01-01T00:30:00Z"} {"asset_id": 1, "latitude": 41.8740, "longitude": -87.6298, "timestamp": "2022-01-01T00:40:00Z"} {"asset_id": 1, "latitude": 41.8800, "longitude": -87.6300, "timestamp": "2022-01-01T00:50:00Z"} {"asset_id": 1, "latitude": 41.8775, "longitude": -87.6244, "timestamp": "2022-01-01T01:00:00Z"} {"asset_id": 1, "latitude": 41.8745, "longitude": -87.6328, "timestamp": "2022-01-01T01:10:00Z"} {"asset_id": 1, "latitude": 41.8825, "longitude": -87.6298, "timestamp": "2022-01-01T01:20:00Z"} {"asset_id": 1, "latitude": 41.8800, "longitude": -87.6250, "timestamp": "2022-01-01T01:30:00Z"}
当您输入这些消息时，您应该看到应用程序的输出，指示位置已经更新。
为了完整起见，还有一个命令需要您注意：在输入这些消息时，您可能会犯错。一个无效的消息可能会导致您的消费者出现问题。一个可能的解决方案是删除主题。删除主题可能听起来很夸张，但这将解决问题。因此，这就是那个命令：kafka-topics --delete --topic asset_location --bootstrap-server localhost:9092.
在本节中，我们已经添加了实时跟踪我们的 资产 在 Apache Kafka 中的位置变化的能力。在本章的最后一节中，我们将使用 Copilot Chat 扩展能力，通过实时监控我们的 资产，尝试确定它们是否位于它们应该位的位置。同样，我们将探讨使用 Spark 和 Kafka 一起完成此分析。完成后，我们将感谢我们的信息安全团队，他们担心我们的核心业务和知识产权过多存在于这些 资产 中。
5.3 使用 Apache Spark 进行分析、学习和跟踪
资产的实时跟踪是一个业务关键功能。您的 IT 资产包含着敏感的商业数据、客户名单、销售数据、利润和损失（PnL）预测以及销售策略，以及许多其他项目。一次丢失的资产可能是公司的生存危机。因此，对于许多信息安全专业人员来说，仔细的管理和监控是首要任务。在本节中，我们旨在使他们的工作变得更加轻松。现代数据平台使实时跟踪资产并在出现可疑情况时发送通知变得微不足道。让我们开始吧。
Apache Spark 是一个强大的、开源的数据处理引擎，以速度、易用性和复杂的分析而闻名。它旨在提供一个比 MapReduce 更好的处理大数据集的替代方案，并且可以处理批处理和实时分析。Spark 提供了 Scala、Java、Python 和 R 的 API，并且具有用于 SQL 查询的内置模块。其核心数据结构，弹性分布式数据集（RDD），支持容错操作，并允许数据在计算机集群上并行处理。
Spark 还包括几个库来扩展其功能，包括用于机器学习的 MLlib，用于处理实时数据流的 Spark Streaming，以及用于处理结构化数据的 Spark SQL 和 DataFrames。这些工具使其非常适合从机器学习到实时数据流和批处理的任务。其内存处理能力使 Spark 比其前身快得多，使其成为大数据处理的热门选择。
首先，我们将要求 Copilot Chat 推荐一种使用 Apache Spark 跟踪我们资产的策略。
清单 5.20 请求 Copilot Chat 如何最好地实时跟踪我们的资产
Let's imagine that our company is in Chicago. We want to create a class called AssetLocationSparkAdapter that will continuously stream AssetLocation json messages from a Kafka topic called asset_location. An AssetLocation has the following attributes asset_id: int, latitude: float, longitude: float, and timestamp: datetime. The AssetLocations are stored in a Postgres database in a table called itam.asset_locations. AssetLocationSparkAdapter should calculate if the new AssetLocation is more than 25 miles from Chicago. If it is, then it should write a message to the console. This should be using the latest version of Spark. How would we do this?
Copilot Chat 生成了应该放入 infrastructure 包中名为 asset_location_spark_adapter.py 的文件的类。幸运的是，它还为每一行包括了注释，因此你应该会发现生成的代码很容易跟踪。导入语句包括 Spark 库以及 geopy。
清单 5.21 运行 Spark 所需的导入项
from pyspark.sql import SparkSession from pyspark.sql.functions import from_json, col, expr from pyspark.sql.types import StructType, StructField, IntegerType, DoubleType, TimestampType from geopy.distance import distance
课程以一个过于臃肿的构造函数开始，该构造函数定义了 Spark 在将 JSON 转换为 DataFrame 时将使用的模式。
注意
如定义的 AssetLocationSparkAdapter 是一个阻塞进程。因此，在 Spark 进程被终止之前，你的 FastAPI 应用程序将不会“完全”启动。你希望这是一个独立的进程，或者你需要引入一个异步框架来使这两个进程同时运行。
接下来，它将启动一个本地 Spark 实例/会话，允许 Spark 连接到 Kafka 主题并持续地流式处理记录。
清单 5.22 负责处理 Kafka 主题并生成通知的 AssessLocationSparkAdapter
class AssetLocationSparkAdapter: def __init__(self): # Define the schema for the incoming JSON data self.schema = StructType([ StructField("asset_id", IntegerType()), StructField("latitude", DoubleType()), StructField("longitude", DoubleType()), StructField("timestamp", TimestampType()) ]) # Create a SparkSession self.spark = SparkSession.builder \ .appName("AssetLocationSparkAdapter") \ .getOrCreate() # Create a streaming DataFrame from the asset_location topic self.df = self.spark \ .readStream \ .format("kafka") \ .option("kafka.bootstrap.servers", "localhost:9092") \ .option("subscribe", "asset_location") \ .option("startingOffsets", "earliest") \ .load() \ .selectExpr("CAST(value AS STRING)") # Parse the incoming JSON data self.parsed_stream = self.df \ .select(from_json(col("value"), self.schema).alias("data")) \ .select("data.*")
AssetLocationSparkAdapter 类的最后一部分将计算资产当前位置到芝加哥的距离。如果差距大于 25 英里，则将结果集发送到控制台。此外，它还提供了一个方法来启动和停止适配器。
清单 5.23 AssessLocationSparkAdapter 计算当前资产位置与芝加哥的距离
# Calculate the distance between the current location and Chicago for each asset self.distance = self.parsed_stream \ .withColumn("distance", expr("calculate_distance(latitude, longitude, 41.8781, -87.6298)")) \ .select(col("asset_id"), col("timestamp"), col("distance")) \ .filter(col("distance") > 25) # Write the results to the console self.query = self.distance \ .writeStream \ .outputMode("append") \ .format("console") \ .start() def run(self): # Start the streaming query self.query.awaitTermination() def stop(self): # Stop the streaming query and SparkSession self.query.stop() self.spark.stop()
calculate_distance 方法接受资产位置的经度和纬度，并使用 geopy.distance 函数确定距离芝加哥的距离。
清单 5.24 Spark 用于计算芝加哥和你的资产之间距离的函数
def calculate_distance(lat1, lon1, lat2, lon2): return distance((lat1, lon1), (lat2, lon2)).miles
在这种情况下，Copilot Chat 生成的代码存在一些问题，阻止其在本地运行。在本地运行并遇到这些问题后，搜索 Stack Overflow，你会发现代码的两个主要问题的解决方案：缺少用于本地运行的环境变量，以及未能注册您的 UDF（用户定义的函数）。幸运的是，你不需要进行测试和研究，因为清单 5.23 中提供了解决方案。
清单 5.25 在本地运行应用程序所需的编辑
os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-10_2.12:3.2.0,org.apache.spark:spark-sql-kafka-0-10_2.12:3.2.0 pyspark-shell' class AssetLocationSparkAdapter: def __init__(self): # Create a SparkSession self.spark = SparkSession.builder \ .appName("AssetLocationSparkAdapter") \ .getOrCreate() self.spark.udf.register("calculate_distance", calculate_distance)
最后，要运行你的 Spark 应用程序，你需要在 main.py 中的 main 函数中更新以下代码。
列表 5.26 对 main 函数的更新
if __name__ == "__main__": adapter = AssetLocationSparkAdapter() adapter.run()
当你将资产位置输入到 Kafka 控制台生产者中时，如果距离芝加哥市中心超过二十五英里，你会注意到条目被写入到控制台中。更新类以将这些结果输出到 Twilio 的短信 API 或类似 SendGrid 的电子邮件服务是微不足道的。
列表 5.27 你的资产位置的流式输出
+--------+-------------------+------------------+ |asset_id| timestamp| distance| +--------+-------------------+------------------+ | 1|2021-12-31 20:30:00| 712.8314662207446| +--------+-------------------+------------------+
恭喜，你正在实时追踪你的资产，并在公司资源离开的时候发送实时警报。
5.4 总结
- GitHub Copilot Chat 是一款创新性工具，结合了 ChatGPT 的全面语言理解和 Copilot 的便捷功能。这是编程辅助领域的一个值得注意的发展，特别是在实时提供详细且上下文相关的建议方面，促进了更高效的编码体验。
- 中介者设计模式是一种独特的行为模式，它促进了对象之间高度解耦，从而增强了代码的模块化。通过在一个中介者对象中包含对象之间的交互，对象可以间接地进行通信，从而减少了依赖性，促进了代码的可重用性和易修改性。
- Apache Kafka 是一个健壮的、分布式的流平台，专为创建实时数据管道和流应用而设计。它可以有效处理来自多种来源的数据流，并将其传输给各种消费者，使其成为处理大量实时或准实时数据的理想解决方案。需要记住的是，Kafka 优化了追加式、不可变数据，而不适用于需要记录更新或删除，或复杂查询的用例。
- Apache Spark 是一款性能卓越的、分布式的数据处理引擎，以其速度、易用性和高级分析功能而闻名。它非常适用于需要实时数据处理或对大量数据集进行操作的场景。然而，对于诸如基本分析或简单聚合等较简单的任务，传统的关系型数据库可能是更合适的选择。
- 尽管生成式人工智能迅速发展，但并非无懈可击。仔细审查所有生成的输出以确保其符合你的特定要求和质量标准至关重要。虽然生成式人工智能不能替代深入的领域知识或编码专业知识，但它通过提供有价值的见解和减少在例行任务上花费的时间来显著提高生产力。
AI 驱动的开发者（MEAP）（二）（3）https://developer.aliyun.com/article/1516331
