【免费下载链接】beamApache Beam is a unified programming model for Batch and Streaming data processing.项目地址https://gitcode.com/gh_mirrors/beam18/beam点击查看免费下载导读本文基于 Apache Beam 官方文档 Anomaly Detection 及其在仓库中的完整配套示例代码展开讲解如何用 Beam 搭建一条实时流式异常检测管线从 Google Cloud Pub/Sub 实时读取 Twitter 文本用 Transformer 语言模型将文本转换为向量嵌入再交给预训练的 HDBSCAN 聚类模型判断每条消息是否为异常离群点。读完本文你将掌握 BeamRunInference在文本嵌入与聚类推理中的用法、ModelHandler的定制方式以及如何把预测结果落地 BigQuery 并触发邮件告警——整套方案可在本地DirectRunner与 Google Cloud DataflowDataflowRunner两种模式下运行。示例代码全部位于仓库 sdks/python/apache_beam/examples/inference/anomaly_detection/包含两条独立管线write_data_to_pubsub_pipeline/数据写入与anomaly_detection_pipeline/流式检测。一、方案总览两条管线协同完成流式异常检测整套方案由两条 Beam 流式管线组成形成数据写入 → 实时检测的闭环write_data_to_pubsub_pipeline加载 Hugging Faceemotion数据集为每条文本分配唯一 ID并写入 Pub/Sub Topic为检测管线提供实时数据流。anomaly_detection_pipeline订阅 Pub/Sub 消息将文本 Token 化后通过RunInference送入 Transformer 语言模型得到 768 维嵌入向量归一化后交给预训练的 HDBSCAN 模型预测最后将结果写入 BigQuery并在检测到异常聚类标签为-1时发送邮件告警。两条管线的核心入口分别为write_data_to_pubsub_pipeline/main.pyanomaly_detection_pipeline/main.py二、数据集Hugging Face emotion20,000 条英文 Twitter 消息示例使用的emotion数据集包含20,000 条英文 Twitter 消息标注了 6 种基本情绪anger愤怒、fear恐惧、joy喜悦、love爱、sadness悲伤、surprise惊讶。数据集分为三个切分train训练、validation验证、test测试因为同时包含文本与类别标签属于监督数据集。以下是train切分中的示例来自官方文档仓库中get_dataset实现与之对应见 pipeline/utils.pyTextType of emotionim grabbing a minute to post i feel greedy wrongAngeri am ever feeling nostalgic about the fireplace i will know that it is still on the propertyLoveive been taking or milligrams or times recommended amount and ive fallen asleep a lot faster but i also feel like so funnyFearon a boat trip to denmarkJoyi feel you know basically like a fake in the realm of science fictionSadnessi began having them several times a week feeling tortured by the hallucinations moving people and figures sounds and vibrationsFear写入管线中的get_dataset(categories, splittrain)函数负责加载数据它按[sadness, joy, love, anger, fear, surprise]顺序建立标签映射并调用load_dataset(emotion, download_modeforce_redownload)获取数据。为了简化演示示例只取其中三类test_categories [sadness, anger, surprise]而非全部六类。三、核心算法HDBSCAN 层次密度聚类示例的异常检测算法选用HDBSCANHierarchical Density-Based Spatial Clustering of Applications with Noise。HDBSCAN 是 DBSCAN 的扩展它将 DBSCAN 转化为层次聚类算法再依据聚类的**稳定性stability**提取出扁平聚类结果。HDBSCAN 对预测的定义非常契合异常检测场景若新数据点被判为离群点/异常模型输出标签-1否则输出其所属的既有聚类簇编号。正是利用输出-1即异常这一特性示例管线在拿到 HDBSCAN 预测结果后即可直接判断某条 Twitter 消息是否为异常并触发后续的告警与入库逻辑。四、管线一将数据写入 Pub/Sub4.1 文件结构与职责写入管线位于 write_data_to_pubsub_pipeline/结构与职责如下write_data_to_pubsub_pipeline/ ├── pipeline/ │ ├── __init__.py │ ├── options.py # 配置 Dataflow 管线选项 │ └── utils.py # 加载 emotion 数据集 两个数据转换 DoFn ├── __init__.py ├── config.py # 全局变量PROJECT_ID、NUM_WORKERS 等 ├── main.py # 管线主体代码 └── setup.py # 定义管线运行所需的包与依赖各文件与文档描述一一对应pipeline/utils.py包含get_dataset加载数据集与AssignUniqueID、ConvertToPubSubMessage两个beam.DoFnpipeline/options.py定义get_pipeline_options根据mode参数选择DirectRunner本地或DataflowRunner云端config.py集中定义PROJECT_ID、REGION、TOPIC_ID、JOB_NAME、NUM_WORKERS等复用的全局变量setup.py声明apache-beam[gcp]2.41.0与datasets2.4.0依赖main.py组装管线。4.2 运行方式运行前需安装依赖并拥有可用的 Google Cloud 项目同时在config.py中配置PROJECT_ID、REGION、TOPIC_ID等变量。# 1. 本地运行DirectRunner python main.py # 2. 在 GCP 上以 Dataflow 运行 python main.py --mode cloud命令行还支持通过-p/--project指定 GCP 项目默认取config.py中的PROJECT_ID。4.3 四个核心转换步骤main.py中的管线主体由四个转换组成源码见 main.pywith beam.Pipeline(optionspipeline_options) as pipeline: docs ( pipeline | Load Documents beam.Create(test_data) | Assign unique key beam.ParDo(AssignUniqueID()) ) _ ( docs | Convert to PubSub Message beam.ParDo(ConvertToPubSubMessage()) | Write to PubSub WriteToPubSub(topiccfg.TOPIC_ID, with_attributesTrue) )加载 emotion 数据集通过get_dataset([sadness, anger, surprise])从 Hugging Face 加载数据为简化只取三个类别分配唯一标识 UIDAssignUniqueID用uuid.uuid4()为每条文本生成唯一 ID产出{id: uid, text: element}字典转换为 Pub/Sub 消息格式ConvertToPubSubMessage构造PubsubMessage(datatext.encode(utf-8), attributes{id: uid})将文本放入消息体、UID 放入 attributes写入 Pub/SubWriteToPubSub(topiccfg.TOPIC_ID, with_attributesTrue)将消息写入指定 Topicwith_attributesTrue表示同时保留 attributes 元数据。写入管线的config.py关键配置如下PROJECT_ID # 需填写你的 GCP 项目 ID REGION us-central1 # 区域 TOPIC_ID fprojects/{PROJECT_ID}/topics/newsgroup-dataset JOB_NAME write-to-pub-sub NUM_WORKERS 1五、管线二流式异常检测5.1 文件结构与职责检测管线位于 anomaly_detection_pipeline/结构与职责如下anomaly_detection_pipeline/ ├── pipeline/ │ ├── __init__.py │ ├── options.py # 配置 Dataflow 管线选项 │ └── transformations.py # 各 DoFn 与工具函数、定制 ModelHandler ├── __init__.py ├── config.py # 全局变量PROJECT_ID、NUM_WORKERS、模型路径等 ├── main.py # 管线主体代码 └── setup.py # 定义管线运行所需的包与依赖关键文件pipeline/transformations.py包含tokenize_sentence、ModelWrapper、CustomSklearnModelHandlerNumpy、NormalizeEmbedding、DecodePubSubMessage、DecodePrediction、TriggerEmailAlertpipeline/options.py与写入管线相同的get_pipeline_options模式config.py集中配置订阅、模型路径、BigQuery 表、告警邮箱等setup.py声明apache-beam[gcp]、hdbscan、scikit-learn、transformers、torch、yagmail等依赖main.py组装完整检测管线。5.2 运行前提与方式运行前需安装依赖拥有 Google Cloud 项目先在config.py中配置PROJECT_ID、REGION、SUBSCRIPTION_ID等变量并先运行写入管线把数据推送到 Pub/Sub。# 1. 本地运行DirectRunner python main.py # 2. 在 GCP 上以 Dataflow 运行 python main.py --mode cloud另一个前提是需要先在训练集上训练好 HDBSCAN 聚类模型并以 joblib 格式保存到config.py中CLUSTERING_MODEL_PATH指定的位置。5.3 八个步骤完整流程检测管线在 main.py 中依次完成以下步骤从 Pub/Sub 读取消息将 Pub/Sub 消息解码为字典key 为 UIDvalue 为 Twitter 文本用 Tokenizer 将文本编码为 Transformer 可读的 token ID 整数序列用RunInference从 Transformer 语言模型获取向量嵌入对嵌入向量做归一化用RunInference从训练好的 HDBSCAN 聚类模型获取异常预测将预测写入 BigQuery以便必要时用新数据重新训练聚类模型检测到异常时发送邮件告警。前两步的代码docs ( pipeline | Read from PubSub ReadFromPubSub(subscriptioncfg.SUBSCRIPTION_ID, with_attributesTrue) | Decode PubSubMessage beam.ParDo(DecodePubSubMessage()) )DecodePubSubMessage从消息体解码文本、从 attributes 取出idyield { text: element.data.decode(utf-8), id: element.attributes[id], }5.4 关键配置项config.py检测管线的 config.py 集中了全部运行参数是部署前必须逐一核对的部分PROJECT_ID REGION us-central1 # 订阅 Pub/Sub Topic 的 Subscription SUBSCRIPTION_ID fprojects/{PROJECT_ID}/subscriptions/newsgroup-dataset-subscription JOB_NAME anomaly-detection-hdbscan NUM_WORKERS 1 EMAIL_ADDRESS xyzgmail.com # 告警接收邮箱 TABLE_SCHEMA { fields: [ {name: text, type: STRING, mode: NULLABLE}, {name: id, type: STRING, mode: NULLABLE}, {name: cluster, type: INTEGER, mode: NULLABLE}, ] } TABLE_URI f{PROJECT_ID}:deliverables_ml6.anomaly-detection MODEL_NAME sentence-transformers-stsb-distilbert-base TOKENIZER_NAME sentence-transformers/stsb-distilbert-base MODEL_STATE_DICT_PATH fgs://{PROJECT_ID}-ml-examples/{MODEL_NAME}/pytorch_model.bin MODEL_CONFIG_PATH TOKENIZER_NAME CLUSTERING_MODEL_PATH fgs://{PROJECT_ID}-ml-examples/anomaly-detection/clustering.joblib各参数含义参数说明PROJECT_IDGCP 项目 ID需要手动填写REGIONDataflow 运行区域默认us-central1SUBSCRIPTION_ID读取消息所用的 Pub/Sub 订阅完整路径格式为projects/{PROJECT_ID}/subscriptions/...JOB_NAMEDataflow 作业名运行时还会追加时间戳避免冲突NUM_WORKERSDataflow 工作节点数默认1EMAIL_ADDRESS异常告警邮件的接收地址TABLE_SCHEMA/TABLE_URIBigQuery 结果表结构与完整表 URIMODEL_STATE_DICT_PATHDistilBERT 模型权重PyTorch state dict的 GCS 路径TOKENIZER_NAME/MODEL_CONFIG_PATHHugging Face 模型仓库名用于加载 Tokenizer 与模型配置CLUSTERING_MODEL_PATH训练好的 HDBSCAN 模型joblib 格式的 GCS 路径5.5 依赖清单setup.py检测管线的 setup.py 声明了以下精确版本依赖REQUIREMENTS [ apache-beam[gcp]2.41.0, hdbscan0.8.28, scikit-learn1.0.2, transformers4.36.0, torch1.13.1, pandas1.3.5, yagmail0.15.283, ]其中apache-beam[gcp]提供 Pub/Sub、BigQuery 与 Dataflow 支持hdbscan提供聚类模型transformerstorch用于语言模型嵌入yagmail用于发送告警邮件。该文件同时被 Dataflow 的--setup_file见 options.py 中setup_file: ./setup.py引用用于在云端 Worker 上安装这些依赖。六、文本嵌入Tokenizer RunInference 定制 PyTorch Handler6.1 为什么需要嵌入与 Token 化要对文本做聚类首先要把文本映射为适合统计分析的数字向量。示例选用 Transformer 语言模型sentence-transformers/stsb-distilbert-base它把句子和段落映射到768 维的稠密向量空间可用于聚类与语义搜索等任务。由于语言模型期望的是 Token 化输入而非原始文本需要先做 **Token 化Tokenization**这一预处理步骤把文本转换为模型可读的 token ID 整数序列。相关代码见 pipeline/transformations.pyTokenizer AutoTokenizer.from_pretrained(cfg.TOKENIZER_NAME) def tokenize_sentence(input_dict): text, uid input_dict[text], input_dict[id] tokens Tokenizer([text], paddingTrue, truncationTrue, return_tensorspt) tokens {key: torch.squeeze(val) for key, val in tokens.items()} return (text, uid), tokenstokenize_sentence接收包含text和id的字典用AutoTokenizer对文本做填充padding与截断truncation处理返回((text, uid), tokens)元组——文本与 ID 作为 key 随 token 一起传递保证推理结果能对回原文。管线中的接入方式normalized_embedding ( docs | Tokenize Text beam.Map(tokenize_sentence) ... )6.2 用 RunInference 获取嵌入Token 化输出随后交给语言模型获取嵌入这一步使用 Beam 的RunInferencePTransform| Get Embedding RunInference(KeyedModelHandler(embedding_model_handler))在 Beam 中RunInference接收一个ModelHandler把一个示例 PCollection转换为预测 PCollection——每个输出是包含输入示例与推理结果的PredictionResult。它内部会基于ModelHandler.batch_elements_kwargs()的配置尝试用beam.BatchElements对示例做批处理inference_args用于传递模型推理所需的额外参数。由于这里的输入是((text, uid), tokens)这种key 示例结构需要使用KeyedModelHandler包装底层 handler。从源码看base.pyKeyedModelHandler专门处理带 key 的示例将PCollection[Tuple[K, E]]转换为PCollection[Tuple[K, P]]从而在预测时保留 key 以关联输入与输出。embedding_model_handler的定义如下见 main.pyembedding_model_handler PytorchNoBatchModelHandler( state_dict_pathcfg.MODEL_STATE_DICT_PATH, model_classModelWrapper, model_params{ config: AutoConfig.from_pretrained(cfg.MODEL_CONFIG_PATH) }, devicecpu, )6.3 PytorchNoBatchModelHandler限制批大小为 1示例自定义了PytorchNoBatchModelHandler继承PytorchModelHandlerKeyedTensor其目的是把批大小限制为 1class PytorchNoBatchModelHandler(PytorchModelHandlerKeyedTensor): Wrapper to PytorchModelHandler to limit batch size to 1. The tokenized strings generated from BertTokenizer may have different lengths, which doesnt work with torch.stack() in current RunInference implementation since stack() requires tensors to be the same size. Restricting max_batch_size to 1 means there is only 1 example per batch in the run_inference() call. def batch_elements_kwargs(self): return {max_batch_size: 1}从代码注释可以理解其技术动机BertTokenizer 生成的 token 序列长度不一而当前RunInference的实现使用torch.stack()拼接批次时要求张量尺寸一致因此将max_batch_size限制为 1保证每次run_inference()调用批次中只有一条示例从根源上规避长度不一致问题。6.4 ModelWrapper让 DistilBertModel 返回嵌入DistilBertModel的forward()默认不直接返回句子嵌入返回的是BaseModelOutput等结构因此示例自定义了ModelWrapper继承DistilBertModel重写forward()并用mean pooling提取句子向量见 transformations.pyclass ModelWrapper(DistilBertModel): Wrapper to DistilBertModel to get embeddings when calling forward function. def forward(self, **kwargs): output super().forward(**kwargs) sentence_embedding ( self.mean_pooling(output, kwargs[attention_mask]).detach().cpu().numpy()) return sentence_embedding # Mean Pooling - Take attention mask into account for correct averaging def mean_pooling(self, model_output, attention_mask): token_embeddings model_output[0] # 第一个元素包含所有 token 嵌入 input_mask_expanded ( attention_mask.unsqueeze(-1).expand(token_embeddings.size()).float()) return torch.sum(token_embeddings * input_mask_expanded, 1) / torch.clamp( input_mask_expanded.sum(1), min1e-9)Mean pooling 的原理模型输出的第一个元素是所有 token 的嵌入矩阵mean_pooling将attention_mask扩展为与嵌入同形状通过掩码区分真实 token 与 padding token仅对真实 token 的嵌入求均值从而得到整句的语义向量同时除以clamp(..., min1e-9)避免除零。得到嵌入后再调用.detach().cpu().numpy()将其转为 NumPy 数组供下游处理。6.5 嵌入归一化由于训练好的聚类模型期望输入是归一化嵌入示例在获取嵌入后立刻做 L2 归一化见 transformations.pyclass NormalizeEmbedding(beam.DoFn): def process(self, element, *args, **kwargs): (text, uid), prediction element embedding prediction.inference l2_norm np.linalg.norm(embedding) yield {text: text, id: uid, embedding: embedding / l2_norm}RunInference输出的是PredictionResult其中inference字段即模型推理结果此处为嵌入向量。NormalizeEmbedding取出该向量除以 L2 范数完成归一化并把结果整理成{text, id, embedding}字典继续传递。| Normalize Embedding beam.ParDo(NormalizeEmbedding())七、异常预测定制 Sklearn Handler 调用 approximate_predict7.1 接入 HDBSCAN 模型归一化后的嵌入被送入训练好的 HDBSCAN 模型做预测predictions ( normalized_embedding | Get Prediction from Clustering Model RunInference(model_handlerclustering_model_handler) )clustering_model_handler的定义见 main.pyclustering_model_handler KeyedModelHandler( CustomSklearnModelHandlerNumpy( model_uricfg.CLUSTERING_MODEL_PATH, model_file_typeModelFileType.JOBLIB ) )这里同样使用KeyedModelHandler包装保证预测结果能按 key文本 UID对回输入。model_file_typeModelFileType.JOBLIB表示模型文件以joblib 序列化格式加载仓库中 sklearn_inference.py 定义了ModelFileType枚举并提供了joblib.load(file)的加载实现。7.2 CustomSklearnModelHandlerNumpy批大小 1 approximate_predict示例基于SklearnModelHandlerNumpy定制了CustomSklearnModelHandlerNumpy做了两件事见 transformations.pyclass CustomSklearnModelHandlerNumpy(SklearnModelHandlerNumpy): # limit batch size to 1 can be removed once: BEAM issue #21863 is fixed def batch_elements_kwargs(self): Limit batch size to 1 for inference return {max_batch_size: 1} # run_inference can be removed once: BEAM issue #22572 is fixed def run_inference(self, batch, model, inference_argsNone): Runs inferences on a batch of numpy arrays. _validate_inference_args(inference_args) vectorized_batch np.vstack(batch) predictions hdbscan.approximate_predict(model, vectorized_batch) return [PredictionResult(x, y) for x, y in zip(batch, predictions)]限制批大小为 1与PytorchNoBatchModelHandler同理batch_elements_kwargs()返回{max_batch_size: 1}覆盖run_inference默认的 sklearn 推理调用的是model.predict()但 HDBSCAN 的predict()只能给已属于某个簇的点打分对新样本需要hdbscan.approximate_predict()才能得到新样本属于哪个簇 / 是否为离群点的近似预测。因此这里用np.vstack(batch)把单条嵌入组装成数组调用hdbscan.approximate_predict(model, vectorized_batch)得到预测再以PredictionResult形式逐条返回。7.3 预测结果解码DecodePrediction把RunInference的输出解码为字典见 transformations.pyclass DecodePrediction(beam.DoFn): def process(self, element): (text, uid), prediction element cluster prediction.inference.item() bq_dict {text: text, id: uid, cluster: cluster} yield bq_dictprediction.inference.item()取出聚类标签整数。若标签为-1即代表异常。八、结果落地BigQuery 写入与邮件告警8.1 写入 BigQuery解码后的预测字典写入 BigQuery 表供后续分析与聚类模型重训练使用_ ( predictions | Decode Prediction beam.ParDo(DecodePrediction()) | Write to BQ beam.io.WriteToBigQuery( tablecfg.TABLE_URI, schemacfg.TABLE_SCHEMA, write_dispositionbeam.io.BigQueryDisposition.WRITE_APPEND, create_dispositionbeam.io.BigQueryDisposition.CREATE_IF_NEEDED, ) )write_dispositionWRITE_APPEND追加写入不覆盖已有数据create_dispositionCREATE_IF_NEEDED表不存在时自动创建schema使用config.py中定义的TABLE_SCHEMAtextSTRING、idSTRING、clusterINTEGER 三列。8.2 邮件告警yagmail若预测结果为异常cluster -1TriggerEmailAlert通过yagmail发送告警邮件见 transformations.pyclass TriggerEmailAlert(beam.DoFn): def setup(self): with open(./cred.json) as json_file: cred json.load(json_file) self.yag_smtp_client yagmail.SMTP(**cred) def process(self, element): (text, uid), prediction element cluster prediction.inference.item() if cluster -1: body fTweet-Id is {uid} and text is {text} self.yag_smtp_client.send( tocfg.EMAIL_ADDRESS, subjectAnomaly Detected, contentsbody, )注意两点实现细节setup()中从当前目录的cred.json读取 SMTP 凭据并初始化yagmail.SMTP客户端因此运行前需要准备该凭据文件邮件仅在cluster -1时发送主题为Anomaly Detected正文包含异常的 Tweet ID 与文本内容。九、运行与排查要点总结综合两条管线落地运行时建议按以下顺序检查配置在两个config.py中填写PROJECT_ID确认TOPIC_ID/SUBSCRIPTION_ID/TABLE_URI/ 模型路径均指向正确的 GCS 与 Pub/Sub 资源检测管线还需确认EMAIL_ADDRESS与cred.json就绪。模型准备提前在 emotion 数据集的train切分上训练 HDBSCAN 模型归一化嵌入后用 joblib 保存至CLUSTERING_MODEL_PATHDistilBERT 权重pytorch_model.bin需上传到MODEL_STATE_DICT_PATH指定的 GCS 位置。先写入、后检测先运行write_data_to_pubsub_pipelinepython main.py或--mode cloud把数据推入 Pub/Sub再启动anomaly_detection_pipeline消费消息。Runner 选择本地验证用默认的 DirectRunner生产/流式场景用python main.py --mode cloud切到 DataflowRunnerget_pipeline_optionsoptions.py会自动设置staging_location、temp_location、setup_file与streamingTrue。批大小限制的由来两处max_batch_size1均是规避当前RunInference批处理对变长张量的限制源码注释中分别关联 Beam issue #21863 与 #22572后续版本可能移除但当前仓库代码中保留该处理以保证正确运行。十、关联资料与延伸阅读官方文档website/www/site/content/en/documentation/ml/anomaly-detection.md写入管线源码write_data_to_pubsub_pipeline/检测管线源码anomaly_detection_pipeline/RunInference与KeyedModelHandler实现sdks/python/apache_beam/ml/inference/base.pysklearn 推理 handlerModelFileType、SklearnModelHandlerNumpysdks/python/apache_beam/ml/inference/sklearn_inference.py同源的在线聚类示例BIRCH 增量聚类便于对照 HDBSCAN 之外的流式聚类方案online-clustering 文档 与 examples/online_clustering.py相关示例目录sdks/python/apache_beam/examples/inference/赞分享【免费下载链接】beamApache Beam is a unified programming model for Batch and Streaming data processing.项目地址https://gitcode.com/gh_mirrors/beam18/beam点击查看免费下载相关推荐Apache Beam 实时文本异常检测实战基于 Pub/Sub、HDBSCAN 与 RunInference 的流式管道Apache Beam 实时文本异常检测实战基于 Pub/Sub、HDBSCAN 与 RunInference 的流式管道 Apache Beam 的异常检测大数据批处理流处理数据工程基于 Apache Beam RunInference 与 HDBSCAN 的实时文本异常检测管道实战指南基于 Apache Beam RunInference 与 HDBSCAN 的实时文本异常检测管道实战指南 本文以 Apache Beam 仓库自带的 anom批处理流处理大数据Apache Beam 流式在线聚类实战Pub/Sub 消息、RunInference 嵌入生成与 BIRCH 状态化聚类Apache Beam 流式在线聚类实战Pub/Sub 消息、RunInference 嵌入生成与 BIRCH 状态化聚类 本文基于 Apache Beam批处理流处理大数据创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考