模块化数字指纹管道（集成训练）简介
动机
本文档介绍了 Morpheus 中数字指纹管道的调整，从现有的基于阶段的方法调整为基于模块的方法；此过程将为理解 23.03 版本中许多新功能的动机和使用示例提供基础。更新后的管道结合了扩展，以促进通过使用控制消息来实现事件驱动的工作流程和人机环路交互。此外，它还引入了一种动态方法来获取和加载数据，进一步增强了管道的功能。
该管道包含一系列互连的模块，旨在创建一个通用的拆分处理管道。这种设计能够接收和处理控制消息，以使用通用或用户特定模型对观察到的事件执行推理等任务。此外，该管道还可以根据聚合或预定义的训练数据训练新模型，从而提供更具适应性和用户友好的体验。
概述
在较高层面，该管道由三个部分组成：前端文件列表加载器，它从 Kafka 主题读取新的控制消息，并将描述的数据源扩展为要处理的数据源，以及处理数据的训练和推理管道。更新后的管道能够接收和处理控制消息，从而可以使用通用或用户特定模型对观察到的事件执行推理等任务。它还允许根据聚合或预定义的训练数据训练新模型。
前端加载器输出一个或多个控制消息，这些消息被传递到训练和推理管道。消息根据消息类型被丢弃或传递到管道中的下一个模块。更新后的管道隐式支持两种不同的工作流程：推理和训练。但是，由于使用了控制消息，它还可以支持混合数据处理工作流程，该工作流程可以处理实时处理的流数据、聚合、批处理，然后用于训练，以及指定要使用的训练数据并能够绕过批处理和聚合阶段的交错式独立训练任务。
此外，更新后的管道还支持人机环路工作流程，例如手动触发针对特定数据集的训练或推理任务的能力，以及实时标记生产推理事件的能力，这些事件可以注入回训练管道中。
以下内容将跟踪在 examples/digital_fingerprinting/production/dfp_integrated_training_streaming_pipeline.py 中声明的管道
# Setup and command line argument parsing
...
# Create an empty pipeline
pipeline = Pipeline(config)
# Create our Kafka source stage that we can read control messages from.
source_stage = pipeline.add_stage(
ControlMessageKafkaSourceStage(config,
bootstrap_servers=kwargs["bootstrap_servers"],
input_topic=list(kwargs["input_topic"]),
group_id=kwargs["group_id"],
poll_interval=kwargs["poll_interval"],
disable_commit=kwargs["disable_commit"],
disable_pre_filtering=kwargs["disable_pre_filtering"]))
# Create our DFP deployment stage that will load our Digital Fingerprinting module into the pipeline.
dfp_deployment_stage = pipeline.add_stage(
MultiPortModulesStage(config,
dfp_deployment_module_config,
input_ports=["input"],
output_ports=["output_0", "output_1"]))
# Connect the source stage to the DFP deployment module
pipeline.add_edge(source_stage, dfp_deployment_stage)
# Run the pipeline
pipeline.run()
有关如何设置和运行 Morpheus 的完整介绍，请参阅入门指南。
来源：python/morpheus_dfp/morpheus_dfp/modules/dfp_deployment.py
这是封装整个数字指纹管道的顶层模块，它主要负责包装训练和推理管道，提供正确的模块接口，并进行一些配置预处理。由于此模块是单体式的，因此它支持大量的配置选项；但是，其中大多数都具有智能默认值，不需要指定。
该模块由三个链式子模块组成
fsspec_dataloader负责将数据源声明扩展为可由管道处理的单个文件。
dfp_training_pipe连接到预处理阶段的输出。负责处理基于训练的控制消息。
dfp_inference_pipe连接到预处理阶段的输出。负责处理基于推理的控制消息。
有关完整参考，请参阅：DFP 部署
@register_module(DFP_DEPLOYMENT, MORPHEUS_MODULE_NAMESPACE)
def dfp_deployment(builder: mrc.Builder):
# Setup and configuration parsing
...
# Make an edge between modules
builder.make_edge(fsspec_dataloader_module.output_port("output"), broadcast)
builder.make_edge(broadcast, dfp_training_pipe_module.input_port("input"))
builder.make_edge(broadcast, dfp_inference_pipe_module.input_port("input"))
out_nodes = [dfp_training_pipe_module.output_port("output"), dfp_inference_pipe_module.output_port("output")]
# Register input port for a module.
builder.register_module_input("input", fsspec_dataloader_module.input_port("input"))
fsspec 数据加载器
来源：morpheus/loaders/fsspec_loader.py
这是新数据加载器模块的一个实例，它利用预定义的 fsspec 样式加载器。该模块用于将 glob 指定的文件列表转换为单个文件路径，并使用这些路径更新控制消息。
有关完整参考，请参阅：数据加载器模块
在训练和推理管道中都使用了许多模块，但可以独立配置。我们将在此处介绍共享模块，然后深入研究每个管道的独特模块。
DFP 预处理
来源：python/morpheus_dfp/morpheus_dfp/modules/dfp_preproc.py
dfp_preproc 模块是 Morpheus 框架中的一个功能组件，它结合了多个与推理和训练相关的数据过滤和处理管道模块。该模块通过将各种模块整合到一个单一的、有凝聚力的单元中来简化管道。dfp_preproc 模块支持配置参数，例如缓存目录、时间戳列名称、预过滤器选项、批处理选项、用户拆分选项以及各种文件类型的支持数据加载器。
该模块本身由一系列链式子模块组成，这些子模块按逻辑顺序连接
filter_control_message_module负责早期过滤不应由管道处理的控制消息。
file_batcher_module负责批处理文件，在封装的训练消息的情况下批处理到单个控制消息中，或者在流数据的情况下批处理到一系列控制消息中。
file_to_df_dataloader_module负责文件检索和插入到 cuDF DataFrame 中。
dfp_split_users_module负责将 DataFrame 拆分为一系列 DataFrame，每个用户一个。
有关完整参考，请参阅：dfp_preproc
@register_module(DFP_PREPROC, MORPHEUS_MODULE_NAMESPACE)
def dfp_preproc(builder: mrc.Builder):
# Setup and configuration parsing
...
# Connect the modules.
builder.make_edge(filter_control_message_module.output_port("output"), file_batcher_module.input_port("input"))
builder.make_edge(file_batcher_module.output_port("output"), file_to_df_dataloader_module.input_port("input"))
builder.make_edge(file_to_df_dataloader_module.output_port("output"), dfp_split_users_module.input_port("input"))
# Register input and output port for a module.
builder.register_module_input("input", filter_control_message_module.input_port("input"))
builder.register_module_output("output", dfp_split_users_module.output_port("output"))
控制消息过滤器
来源：morpheus/modules/filter_control_message.py
filter_control_message 模块是一个组件，旨在根据指定的过滤条件丢弃控制消息。此模块允许用户配置过滤选项，例如任务类型和数据类型。当启用任务过滤或数据类型过滤时，该模块会处理控制消息以验证它们是否满足指定的条件。如果控制消息与条件不匹配，则会将其丢弃。该模块使用节点函数来过滤和处理控制消息，并注册输入和输出端口以促进与数据处理管道的无缝集成。
有关完整参考，请参阅：控制消息过滤器
文件批处理程序
来源：morpheus/modules/file_batcher.py
file_batcher 模块是一个组件，负责加载输入文件，过滤掉早于指定时间窗口的文件，并将剩余的文件按落在时间窗口内的周期进行分组。此模块为参数提供配置，例如批处理选项、缓存目录、文件类型、过滤空值、数据模式和时间戳列名称。file_batcher 模块处理控制消息，验证它们，并生成文件列表及其时间戳。然后，该模块按给定的周期对文件进行分组，为每个批次创建控制消息，并将它们向下游发送以进行进一步处理。节点函数用于处理控制消息，并注册输入和输出端口以将模块无缝集成到数据处理管道中。
文件批处理程序是第一个开始与之前的原始数据管道（23.03 之前的版本）有很大不同的管道组件之一。除了之前的功能之外，文件批处理程序现在可以识别控制消息，并且可以处理流式和封装的控制消息，这是由控制消息元数据的 data_type 属性设置为 streaming 或 payload 来表示的属性。此外，文件批处理程序对于 period、sampling_rate_s、start_time 和 end_time 的默认处理条件现在可以被控制消息的 batching_options 元数据条目中的相应值覆盖。
在流数据的情况下，文件批处理程序将像以前一样运行，按照 period、sampling_rate_s、start_time 和 end_time 属性指定的周期对文件进行分组，为每个批次创建控制消息，并将它们向下游转发。在封装数据的情况下，文件批处理程序将以类似的方式运行，但只会为整个有效负载创建一个控制消息，并将其向下游转发。通过这种方式，可以将所有必要的训练数据附加到给定的训练任务，并跳过任何下游聚合。
有关完整参考，请参阅：文件批处理程序
@register_module(FILE_BATCHER, MORPHEUS_MODULE_NAMESPACE)
def file_batcher(builder: mrc.Builder):
# Setup and configuration parsing
...
文件到 DF 数据加载器
来源：morpheus/loaders/file_to_df_loader.py
这是新数据加载器模块的一个实例，它利用预定义的 file_to_df 样式加载器。该模块用于处理引用需要检索（可能需要缓存），然后加载到 cuDF DataFrame 中的文件的 load 任务，cuDF DataFrame 设置为控制消息有效负载。
有关完整参考，请参阅：数据加载器模块
DFP 拆分用户
来源：python/morpheus_dfp/morpheus_dfp/modules/dfp_split_users.py
dfp_split_users 模块负责根据用户 ID 拆分输入数据。该模块提供配置选项，例如回退用户名、包括通用用户、包括单个用户以及指定要在输出中包含或排除的用户 ID 列表。
该模块通过从消息有效负载中提取用户信息、根据提供的配置过滤数据以及按用户 ID 拆分数据来处理控制消息。对于每个用户 ID，该函数都会生成一个新的控制消息，其中包含相应的数据，并将其向下游发送以进行进一步处理。
有关完整参考，请参阅：DFP 拆分用户
@register_module(DFP_SPLIT_USERS, MORPHEUS_MODULE_NAMESPACE)
def dfp_split_users(builder: mrc.Builder):
# Setup and configuration parsing
...
DFP 滚动窗口
来源：python/morpheus_dfp/morpheus_dfp/modules/dfp_rolling_window.py
dfp_rolling_window 模块负责维护历史数据的滚动窗口，充当流式缓存和批处理系统。该模块提供各种配置选项，例如聚合跨度、缓存目录、缓存选项、时间戳列名称和触发条件。
该模块的主要功能是处理包含数据的控制消息。对于每个控制消息，该函数确定用户 ID 和数据类型，然后尝试使用缓存中的历史数据构建滚动窗口。如果基于触发条件有足够的数据可用，则该函数会返回一个控制消息，其中包含用于进一步处理的适当历史数据。
滚动窗口模块是另一个已更新为可识别控制消息的模块示例。它将区分流式和有效负载控制消息，并相应地处理它们。在流式控制消息的情况下，该模块将像以前一样处理消息，并缓存流式数据并返回，或者在满足触发条件时，返回包含适当历史数据的控制消息。在有效负载控制消息的情况下，将完全跳过滚动窗口模块，而只是将消息转发到下一个阶段。
滚动窗口模块也已更新以支持额外的 batch 运行模式，在该模式下，它将缓存流式数据，直到满足触发条件，生成一个包含所有现有数据的新控制消息，刷新缓存，然后将消息向下游转发。批处理缓存是流式推理管道的默认模式，并通过减少所需的簿记来提高性能。此运行模式由模块配置的 cache_mode 属性表示。
有关完整参考，请参阅：DFP 滚动窗口
@register_module(DFP_ROLLING_WINDOW, MORPHEUS_MODULE_NAMESPACE)
def dfp_rolling_window(builder: mrc.Builder):
# Setup and configuration parsing
...
DFP 数据准备
来源：python/morpheus_dfp/morpheus_dfp/modules/dfp_data_prep.py
dfp_data_prep 模块负责为推理或模型训练准备数据。该模块需要为数据准备定义模式。
该模块的主要功能在 process_features 函数中。对于每个包含数据的控制消息，该函数根据给定的模式处理数据的列。然后，将处理后的 DataFrame 应用于控制消息有效负载。
有关完整参考，请参阅：DFP 数据准备
@register_module(DFP_DATA_PREP, MORPHEUS_MODULE_NAMESPACE)
def dfp_data_prep(builder: mrc.Builder):
# Setup and configuration parsing
...
来源：python/morpheus_dfp/morpheus_dfp/modules/dfp_training_pipe.py
DFP 训练管道模块是一个整合模块，它集成了对训练过程至关重要的多个 DFP 管道模块。此模块功能为训练管道提供了一个单一入口点，简化了模型训练的过程。该模块为管道中各个阶段提供可配置的参数，包括数据批处理、数据预处理和用于模型训练的数据编码。此外，MLflow 模型写入器选项允许保存训练后的模型以供将来使用。
该模块本身由一系列链式子模块组成，每个子模块都在训练中执行特定任务
预处理数据过滤和预处理
dfp_rolling_window数据缓存和批处理
dfp_data_prep数据编码
dfp_training模型训练
mlflow_model_writer将模型和遥测数据保存到 MLflow
有关完整参考，请参阅：DFP 训练管道
@register_module(DFP_TRAINING_PIPE, MORPHEUS_MODULE_NAMESPACE)
def dfp_training_pipe(builder: mrc.Builder):
# Setup and config parsing
...
# Make an edge between the modules.
builder.make_edge(preproc_module.output_port("output"), dfp_rolling_window_module.input_port("input"))
builder.make_edge(dfp_rolling_window_module.output_port("output"), dfp_data_prep_module.input_port("input"))
builder.make_edge(dfp_data_prep_module.output_port("output"), dfp_training_module.input_port("input"))
builder.make_edge(dfp_training_module.output_port("output"), mlflow_model_writer_module.input_port("input"))
# Register input and output port for a module.
builder.register_module_input("input", preproc_module.input_port("input"))
builder.register_module_output("output", mlflow_model_writer_module.output_port("output"))
DFP 训练
来源：python/morpheus_dfp/morpheus_dfp/modules/dfp_training.py
dfp_training 模块功能负责训练模型。on_data 函数被定义为处理传入的 ControlMessage 实例。它从 ControlMessage 中检索用户 ID 和输入数据，使用指定的 model_kwargs 创建 AutoEncoder 类的实例，并在输入数据上训练模型。输出消息包括训练后的模型和元数据。
有关完整参考，请参阅：DFP 训练
@register_module(DFP_TRAINING, MORPHEUS_MODULE_NAMESPACE)
def dfp_inference(builder: mrc.Builder):
# Setup and config parsing
...
MLflow 模型写入器
来源：morpheus/modules/mlflow_model_writer.py
mlflow_model_writer 模块负责将训练后的模型上传到 MLflow 服务器。
对于接收到的每个 ControlMessage（包含训练后的模型），该函数会将模型以及关联的元数据（例如实验名称、运行名称、参数、指标和模型签名）上传到 MLflow。如果 MLflow 服务器在 Databricks 上运行，则该函数还会将所需的权限应用于注册的模型。
有关完整参考，请参阅：MLflow 模型写入器
@register_module(MLFLOW_MODEL_WRITER, MORPHEUS_MODULE_NAMESPACE)
def mlflow_model_writer(builder: mrc.Builder):
# Setup and configuration parsing
...
来源：python/morpheus_dfp/morpheus_dfp/modules/dfp_inference_pipe.py
dfp_inference_pipe 模块功能将与推理过程相关的多个数字指纹管道 (DFP) 模块整合到一个模块中。其目的是通过组合所有必要的组件来简化推理管道的创建和配置。
该模块设置了一系列互连的组件，这些组件处理推理过程的各个阶段，例如预处理、滚动窗口聚合、数据准备、推理、检测过滤、后处理、序列化以及将输出写入文件。
该模块本身由一系列链式子模块组成，每个子模块都在推理管道中执行特定任务
dfp_preproc数据过滤和预处理
dfp_rolling_window数据缓存和批处理
dfp_data_prep数据编码
dfp_inference模型推理
filter_detections检测过滤
dfp_post_proc检测后处理
serialize检测序列化
write_to_file检测结果写入文件
有关完整参考，请参阅：DFP 推理管道
@register_module(DFP_INFERENCE_PIPE, MORPHEUS_MODULE_NAMESPACE)
def dfp_inference_pipe(builder: mrc.Builder):
# Setup and config parsing
...
# Make an edge between the modules.
builder.make_edge(preproc_module.output_port("output"), dfp_rolling_window_module.input_port("input"))
builder.make_edge(dfp_rolling_window_module.output_port("output"), dfp_data_prep_module.input_port("input"))
builder.make_edge(dfp_data_prep_module.output_port("output"), dfp_inference_module.input_port("input"))
builder.make_edge(dfp_inference_module.output_port("output"), filter_detections_module.input_port("input"))
builder.make_edge(filter_detections_module.output_port("output"), dfp_post_proc_module.input_port("input"))
builder.make_edge(dfp_post_proc_module.output_port("output"), serialize_module.input_port("input"))
builder.make_edge(serialize_module.output_port("output"), write_to_file_module.input_port("input"))
# Register input and output port for a module.
builder.register_module_input("input", preproc_module.input_port("input"))
builder.register_module_output("output", write_to_file_module.output_port("output"))
DFP 推理
来源：python/morpheus_dfp/morpheus_dfp/modules/dfp_inference.py
dfp_inference 模块功能创建一个推理模块，该模块检索训练后的模型并对输入数据执行推理。该模块需要在其参数中配置 model_name_formatter 和 fallback_username。
该函数定义了一个 get_model 方法来加载特定用户的模型，以及一个 process_task 方法来处理单个推理任务。process_task 方法检索用户 ID，提取有效负载，并将 DataFrame 转换为 pandas 格式。然后，它尝试加载指定用户 ID 的模型，并使用加载的模型执行推理。最后，它将输入数据中的任何其他列添加到结果 DataFrame，并创建一个包含输出和元数据的消息。
有关完整参考，请参阅：DFP 推理
@register_module(DFP_INFERENCE, MORPHEUS_MODULE_NAMESPACE)
def dfp_inference(builder: mrc.Builder):
# Setup and config parsing
...
过滤检测结果
来源：morpheus/modules/filter_detections.py
filter_detections 模块功能旨在根据指定阈值，基于张量或 DataFrame 列中的值来过滤 DataFrame 中的行。如果指定字段中与其关联的值小于或等于阈值，则排除这些行。
此模块可以在两种模式下运行，由 copy 参数设置。当 copy=True 时，符合过滤条件的行将复制到新的 DataFrame 中。当 copy=False 时，将改用切片视图。
该函数定义了 find_detections 方法来确定过滤器源并识别与过滤条件匹配的行。filter_copy 和 filter_slice 方法负责根据所选模式处理过滤过程。
@register_module(FILTER_DETECTIONS, MORPHEUS_MODULE_NAMESPACE)
def filter_detections(builder: mrc.Builder):
# Setup and config parsing
...
有关完整参考，请参阅：过滤检测结果
DFP 后处理
来源：python/morpheus_dfp/morpheus_dfp/modules/dfp_postprocessing.py
dfp_postprocessing 模块功能对输入数据执行后处理任务。
@register_module(DFP_POST_PROCESSING, MORPHEUS_MODULE_NAMESPACE)
def dfp_postprocessing(builder: mrc.Builder):
# Setup and config parsing
...
有关完整参考，请参阅：DFP 后处理
序列化
来源：morpheus/modules/serialize.py
序列化模块功能负责从 ControlMessage 对象中过滤列并发出 MessageMeta 对象。
convert_to_df 函数将 DataFrame 转换为 JSON 行。它接受一个 ControlMessage 实例、include_columns（要包含的列的模式）、exclude_columns（要排除的列的模式列表）和 columns（要包含的列的列表）。该函数根据包含和排除模式过滤输入 DataFrame 的列，并检索过滤后的列的元数据。
该模块功能将包含和排除模式编译为正则表达式。然后，它使用 convert_to_df 函数以及编译后的包含和排除模式以及指定的列创建一个节点。
@register_module(SERIALIZE, MORPHEUS_MODULE_NAMESPACE)
def serialize(builder: mrc.Builder):
# Setup and config parsing
...
有关完整参考，请参阅：序列化
写入文件
来源：morpheus/modules/write_to_file.py
write_to_file 模块功能将所有消息写入文件。
convert_to_strings 函数接受 DataFrame（pandas 或 cuDF），并根据文件类型（JSON 或 CSV）将其转换为适当的字符串格式。它检查是否包含索引列。
@register_module(WRITE_TO_FILE, MORPHEUS_MODULE_NAMESPACE)
def write_to_file(builder: mrc.Builder):
# Setup and config parsing
...
有关完整参考，请参阅：写入文件
以下是使用示例 Azure 和 Duo 数据集运行模块化 DFP 管道的步骤。
系统要求
Docker 和 Docker Compose 安装在主机上
支持 GPU 和 NVIDIA Container Toolkit
注意：有关 GPU 要求，请参阅入门指南。
构建服务
从 Morpheus 存储库的根目录运行
cd examples/digital_fingerprinting/production
docker compose build
注意：这需要 1.28.0 或更高版本的 Docker Compose，最好是 v2。如果您遇到类似于以下的错误
ERROR: The Compose file './docker-compose.yml' is invalid because:
services.jupyter.deploy.resources.reservations value Additional properties are not allowed ('devices' was
unexpected)
这很可能是由于使用了旧版本的 docker-compose 命令，请使用 docker compose 重新运行构建。有关更多信息，请参阅迁移到 Compose V2。
下载示例数据集
首先，我们需要安装 s3fs，然后运行 examples/digital_fingerprinting/fetch_example_data.py 脚本。这将把示例数据下载到 examples/data/dfp 目录中。
从 Morpheus 存储库运行
pip install s3fs
python examples/digital_fingerprinting/fetch_example_data.py all
运行 Morpheus 管道
从 examples/digital_fingerprinting/production 目录运行
docker compose run morpheus_pipeline bash
要在容器中使用示例数据集运行 DFP 管道，请从 examples/digital_fingerprinting/production/ 运行以下命令
Duo 训练管道
python dfp_integrated_training_batch_pipeline.py \ --log_level DEBUG \ --source duo \ --start_time "2022-08-01" \ --duration "60d" \ --train_users generic \ --input_file "./morpheus/control_messages/duo_payload_training.json"
Duo 推理管道
python dfp_integrated_training_batch_pipeline.py \ --log_level DEBUG \ --source duo \ --start_time "2022-08-30" \ --input_file "./morpheus/control_messages/duo_payload_inference.json"
Duo 训练 + 推理管道
python dfp_integrated_training_batch_pipeline.py \ --log_level DEBUG \ --source duo \ --start_time "2022-08-01" \ --duration "60d" \ --train_users generic \ --input_file "./morpheus/control_messages/duo_payload_load_train_inference.json"
Azure 训练管道
python dfp_integrated_training_batch_pipeline.py \ --log_level DEBUG \ --source azure \ --start_time "2022-08-01" \ --duration "60d" \ --train_users generic \ --input_file "./morpheus/control_messages/azure_payload_training.json"
Azure 推理管道
python dfp_integrated_training_batch_pipeline.py \ --log_level DEBUG \ --source azure \ --start_time "2022-08-30" \ --input_file "./morpheus/control_messages/azure_payload_inference.json"
Azure 训练 + 推理管道
python dfp_integrated_training_batch_pipeline.py \ --log_level DEBUG \ --source azure \ --start_time "2022-08-01" \ --duration "60d" \ --train_users generic \ --input_file "./morpheus/control_messages/azure_payload_load_train_inference.json"
输出字段
输出文件 dfp_detectiions_duo.csv 和 dfp_detections_azure.csv 将包含输入数据集中检测到异常的那些日志；这由 mean_abs_z 字段中的 z 分数确定。默认情况下，z 分数为 2.0 或更高的任何日志都被认为是异常的。请参阅 DFPPostprocessingStage。
运行上述示例生成的输出文件中的大多数字段都是输入字段或从输入字段派生的字段。其他输出字段是
字段 |
类型 |
描述 |
|---|---|---|
event_time |
文本 | ISO 8601 格式的日期字符串，Morpheus 检测到异常的时间 |
model_version |
文本 | 用于执行推理的模型的名称和版本，格式为 <model name>:<version> |
max_abs_z |
浮点数 | 所有特征中的最大 z 分数 |
mean_abs_z |
浮点数 | 所有特征中的平均 z 分数 |
除此之外，对于每个输入特征，都将存在以下输出字段
字段 |
类型 |
描述 |
|---|---|---|
<feature name>_loss |
浮点数 | 损失 |
<feature name>_z_loss |
浮点数 | 损失 z 分数 |
<feature name>_pred |
浮点数 | 预测值 |
有关这些字段的更多信息，请参阅DFPInferenceStage。