探索如何使用 ChatGPT 创建物联网 Kafka 事件消费者，并使用 API 逻辑服务器来生成定义范围之外的温度读取事件。
Apache Kafka 已成为从静态数据（数据库事务）迁移到事件流的企业架构的明显领导者。有许多演示文稿解释了 Kafka 的工作原理以及如何扩展此技术堆栈（本地或云）。使用 ChatGPT 构建一个微服务来消费消息并丰富、转换和持久化是该项目的下一阶段。在此示例中，我们将使用来自 IoT 设备 （RaspberryPi） 的输入，该设备每隔几秒钟发送一次 JSON 温度读数。
使用消息
生成（并记录）每条 Kafka 事件消息时，Kafka 微服务使用者已准备好处理每条消息。我让 ChatGPT 生成一些 Python 代码，它为我提供了从命名的“主题”中轮询和读取的基础知识。我得到的是一个非常好的开始，可以消耗主题、键和 JSON 有效负载。ChatGPT 创建了代码，使用 SQLAlchemy 将其持久化到数据库中。然后，我想转换 JSON 有效负载，并使用 API Logic Server（ALS - GitHub 上的一个开源项目）规则来取消 JSON，验证、计算并根据给定范围之外的源温度生成一组新的消息有效负载。
ChatGPT: “design a Python Event Streaming Kafka Consumer interface”注意：ChatGPT 选择了 Confluent Kafka 库（并使用其 Docker Kafka 容器）- 您可以修改代码以使用其他 Python Kafka 库。
SQLAlchemy模型
使用 API Logic Server（ALS：Python 开源平台），我们连接到 MySQL 数据库。ALS 将读取这些表，并为每个 ORM 端点创建一个 SQLAlchemy ORM 模型、一个 react-admin 用户界面、safrs-JSON Open API （Swagger） 和一个正在运行的 REST Web 服务。新的温度表将包含时间戳、IoT 设备 ID 和温度读数。在这里，我们使用 ALS 命令行实用程序来创建 ORM 模型：
ApiLogicServer create --project_name=iot --db_url=mysql+pymysql://root:password@127.0.0.1:3308/iotAPI Logic Server 生成的类用于保存我们的值。Temperature
class Temperature(SAFRSBase, Base):br __tablename__ = 'Temperature'br _s_collection_name = 'Temperature' # type: ignorebr __bind_key__ = 'None'brbr Id = Column(Integer, primary_key=True)br DeviceId = Column(Integer, nullable=False)br TempReading = Column(Integer, nullable=False)br CreateDT = Column(TIMESTAMP, server_default=text("CURRENT_TIMESTAMP"), nullable=False)br KafkaMessageSent = Column(Booelan, default=text("False"))变化
因此，我们不是将 Kafka JSON 使用者消息再次保存在 SQL 数据库中（并触发规则来执行工作），而是解包 JSON 有效负载 （） 并将其插入 Temperature 表，而不是保存 JSON 有效负载。我们让声明性规则处理每个温度读数。util.row_to_entity
蟒
entity = models.Temperature()br util.row_to_entity(message_data, entity) br session.add(entity)当消费者收到消息时，它会将其添加到会话中，从而触发规则（如下）。commit_event
声明性逻辑：生成消息
使用 API Logic Server（使用 SQLAlchemy、Flask 和类似 LogicBank 电子表格的规则引擎构建的自动化框架：公式、总和、计数、复制、约束、事件等），我们在 ORM 实体上添加一个声明性规则。当每条消息都保存到 Temperature 表中时，将调用该规则。如果温度读数超过或小于 ，我们将发送有关该主题的 Kafka 消息。我们还添加了一个约束，以确保我们在正常范围 （-） 内接收数据。我们将让另一个事件使用者处理警报消息。
commit_eventTemperaturecommit_eventMAX_TEMPMIN_TEMP“TempRangeAlert”32132TDD 行为测试
使用 TDD（Test Driven Development），我们可以编写一个 Behave 测试，将记录直接插入到 Temperature 表中，然后检查返回值。行为以 /（.feature 文件）开头。对于每个场景，我们使用装饰器编写相应的 Python 类。
KafkaMessageSentFeatureScenarioBehave
功能定义
Feature: TDD Temperature ExamplebrbrScenario: Temperature Processingbr Given A Kafka Message Normal (Temperature)br When Transactions normal temperature is submittedbr Then Check KafkaMessageSent Flag is FalsebrbrScenario: Temperature Processingbr Given A Kafka Message Abnormal (Temperature)br When Transactions abnormal temperature is submittedbr Then Check KafkaMessageSent Flag is TrueTDD Python 类
from behave import *brimport safrsbrbrdb = safrs.DB brsession = db.sessionbrbrdef insertTemperature(temp:int) -> bool:br entity = model.Temperature()br entity.TempReading = tempbr entity.DeviceId = 'local_behave_test'br session.add(entity) br return entity.KafkaMessageSent brbr@given('A Kafka Message Normal (Temperature)')brdef step_impl(context):br context.temp = 76br assert Truebrbr@when('Transactions normal temperature is submitted')brdef step_impl(context):br context.response_text = insertTemperature(context.temp)brbr@then('Check KafkaMessageSent Flag is False')brdef step_impl(context):br assert context.response_text == False总结
使用 ChatGPT 为 Consumer 和 Producer 生成 Kafka 消息代码似乎是一个很好的起点。安装 Confluent Docker for Kafka。将 API Logic Server 用于声明性逻辑规则，使我们能够将公式、约束和事件添加到正常的事务流中，并将其添加到我们的 SQL 数据库中，并生成（和转换）新的 Kafka 消息，这是一个很好的组合。ChatGPT 和声明式逻辑是“配对编程”的下一个层次。
from confluent_kafka import Producerbrconf = {'bootstrap.servers': 'localhostd:9092'}brproducer = Producer(conf)brMAX_TEMP = arg.MAX_TEMP or 102brMIN_TEMP = arg.MIN_TTEMP or 78br brdef produce_message(br row: models.KafkaMessage, br old_row: models.KafkaMessage, br logic_row: LogicRow):br br if logic_row.isInserted() and row.TempReading > MAX_TEMP:br produce(topic="TempRangeAlert", br key=row.Id,br value=f"The temperature {row.TempReading}F exceeds {MAX_TEMP}F on Device {row.DeviceId}")br row.KafkaMessageSent = Truebrbr if logic_row.isInserted() and row.TempReading < MIN_TEMP:br produce(topic="TempRangeAlert", br key=row.Id,br value=f"The temperature {row.TempReading}F less than {MIN_TEMP}F on Device {row.DeviceId}")brbr row.KafkaMessageSent = Truebr br Rules.constraint(models.Temperature, br as_expression= lambda row: row.TempReading < 32 or row.TempReading > 132, br error_message= "Temperature {row.TempReading} is out of range"br Rules.commit_event(models.Temperature, calling=produce_message)仅当温度读数大于或小于时才会生成警报消息。Constraint 将在调用 commit 事件之前检查温度范围（请注意，规则始终是无序的，可以随着规范的变化而引入）。MAX_TEMPMIN_TEMP
原文标题：Kafka Event Streaming AI and Automation
原文链接：
https://dzone.com/articles/event-streaming-ai-amp-automation
作者：Tyler Band
编译：LCR