Apache Beam Agent Skills 体系：面向 AI 代理的代码库专业化技能框架

大数据批处理流处理数据工程【免费下载链接】beamApache Beam is a unified programming model for Batch and Streaming data processing.项目地址https://gitcode.com/gh_mirrors/beam4/beam点击查看免费下载本文档解读 Apache Beam 仓库中.agent/skills/目录的设计这是一套专门为 AI Agent智能编程助手打造的、可发现、可激活、可执行的技能Skill集合覆盖 Beam 核心编程模型、Gradle 构建、I/O 连接器、Runner 配置、CI/CD 等 12 个开发领域。读完本文你将掌握这套技能框架的目录结构、YAML 元数据规范、三阶段运作机制以及如何为仓库新增一个高质量技能从而让 Agent 在 Beam 代码库中高效、准确地完成各类专业化任务。一、Agent Skills 是什么Apache Beam 是一个采用统一编程模型处理批流数据Batch and Streaming data processing的开源项目。随着 AI Agent 越来越多地参与代码开发Beam 仓库在.agent/skills/目录下组织了一套Agent Skills——即一组帮助 Agent 在 Apache Beam 代码库中执行专业化任务的指令文件。每个技能的核心是一份SKILL.md指令文件文件头部带有 YAML frontmatter包含name与description元数据正文则是清晰、可执行的指令内容。Agent 通过扫描技能描述来判断某个技能是否与当前任务匹配匹配后加载完整指令并按其执行。二、技能全景仓库内已提供的技能清单.agent/skills/目录目前提供以下技能覆盖了 Beam 开发的各个关键环节技能说明adding-new-metadata如何在WindowedValue中新增并传播新元数据字段避免元数据丢失beam-conceptsBeam 核心编程模型PCollections、PTransforms、windowing、triggersbeam-dofn-modernizer重写 Beam DoFn 方法移除遗留的ProcessContext/OnTimerContext用法ci-cdGitHub Actions 工作流、CI 失败排查、测试触发contributingPR 工作流、issue 管理、代码评审、发布周期gradle-build构建命令、flags、发布、故障排查io-connectors51 个 I/O 连接器、测试模式与用法示例developing-new-io-connectors开发全新 I/O 连接器的详细指南java-developmentJava SDK 开发、构建、测试与项目结构license-compliance所有新文件必须携带 Apache 2.0 许可头python-developmentPython SDK 环境搭建、测试、构建流水线runnersDirect、Dataflow、Flink、Spark runner 配置yaml-developmentYAML SDK 开发、环境搭建、测试与关键概念这些技能彼此呼应、层层递进beam-concepts建立模型认知java-development、python-development、yaml-development覆盖三种语言 SDK 的开发实践io-connectors与developing-new-io-connectors分别讲使用连接器与开发连接器gradle-build与ci-cd解决构建与工程化问题adding-new-metadata和beam-dofn-modernizer则深入核心执行引擎的特定改造场景。三、技能如何工作三阶段运作机制.agent/skills/README.md明确了技能生效的三阶段机制Discovery发现Agent 扫描各技能的描述即 YAML frontmatter 中的description字段筛选出与当前任务相关的技能Activation激活当某个技能与任务匹配时Agent 读取该技能完整的SKILL.md指令文件Execution执行Agent 遵循技能中的指令在代码库中完成任务。这一机制对技能编写者提出了明确要求description必须简洁且描述清楚何时使用该技能concise description for when to use this skill因为它直接决定了技能能否被 Agent 在 Discovery 阶段准确命中。例如beam-concepts的 description 为Explains core Apache Beam programming model concepts including PCollections, PTransforms, Pipelines, and Runners. Use when learning Beam fundamentals or explaining pipeline concepts.——既概括了内容又给出了明确的适用场景提示。四、技能结构规范SKILL.md 与 YAML frontmatter每个技能文件夹内部结构统一SKILL.md—— 主指令文件带 YAML frontmatter。标准模板如下来自.agent/skills/README.md--- name: skill-name description: Concise description for when to use this skill --- # Skill Content Detailed instructions...从仓库内已实现的技能可以看到该规范的实际形态。例如 beam-concepts/SKILL.md 的 frontmatter--- name: beam-concepts description: Explains core Apache Beam programming model concepts including PCollections, PTransforms, Pipelines, and Runners. Use when learning Beam fundamentals or explaining pipeline concepts. ---注意所有SKILL.md的 frontmatter 之前都保留了 Apache License 2.0 的许可头注释这与 license-compliance 技能的要求一致——仓库中所有新文件都必须携带 Apache 2.0 许可头。因此新增技能文件时同样需要先写入标准的许可头再放置 frontmatter。五、从技能内容看 Beam 开发实战要点虽然 README 是技能的目录页但每个技能正文都浓缩了大量可操作的实战内容。以下按主题抽取核心要点便于快速定位与使用。5.1 构建系统与常用命令gradle-buildBeam 是使用 Gradle 构建的单一仓库mono-repoJava、Python、Go、网站全部由同一个 Gradle 工程管理。关键文件包括根目录的settings.gradle.kts模块定义、gradle.properties全局属性与版本以及buildSrc/下的自定义插件BeamModulePlugin统一管理 Java 依赖、配置 Java/Python/Go/Proto/Docker 等工程类型。常用命令速查# 编译指定模块不跑测试 ./gradlew :sdks:java:core:compileJava # 构建并测试指定模块 ./gradlew :sdks:java:core:build # 运行指定测试类 ./gradlew :sdks:java:core:test --tests *MyTest # Java 代码格式化Spotless ./gradlew spotlessApply # 发布到 Maven Local ./gradlew -Ppublishing :sdks:java:core:publishToMavenLocal常用 flags 包括-p path指定工程目录、-x task排除任务、--tests pattern过滤测试、-PdisableSpotlessChecktrue、-PskipCheckerFramework、--continue失败后继续、--scan生成构建扫描等。构建 Docker SDK 容器时若日志出现 No output specified with docker-container driver 警告需加-PuseDockerBuildx。5.2 Java 与 Python SDK 开发java-development / python-developmentJava 侧sdks/java/核心 SDK 位于sdks/java/coreI/O 连接器位于sdks/java/io/扩展位于sdks/java/extensions/runner 实现位于runners/direct-java、flink/、spark/、google-cloud-dataflow-java/。每个 Java 模块的build.gradle都以apply plugin: org.apache.beam.module开头。测试命名约定单元测试*Test.java集成测试*IT.java使用TestPipelineJUnit HTML 报告位于project/build/reports/tests/test/index.html。运行 Dataflow 集成测试可通过-DbeamTestPipelineOptions[--runnerTestDataflowRunner,--project...]传入流水线选项。Python 侧sdks/python/主包为apache_beam/包含transforms/核心转换、io/连接器、ml/RunInference 等、runners/runner 包装与 worker harness。推荐用 pyenv 建虚拟环境后以可编辑模式安装pip install -e .[gcp,test]。单元测试文件命名*_test.py集成测试*_it_test.py配合--test-pipeline-options--runnerTestDataflowRunner ...。运行流水线遇到NameError时是主模块的全局变量未被序列化所致加--save_main_session即可。质量工具链为 rufflint、pyrefly类型检查、yapf格式化、isortimport 排序。5.3 核心编程模型beam-concepts技能正文系统梳理了 Beam 模型的关键抽象Pipeline封装整个数据处理任务读、变换、写PCollection分布式数据集具备不可变、分布式并行、有界batch/无界streaming、每个元素带事件时间戳、元素被分配到窗口等性质PTransform对 PCollection 的数据处理操作核心转换包括ParDo通用并行处理、GroupByKey按键分组、CoGroupByKey多 PCollection 按键连接、Combine求和/均值等聚合支持Combine.globally与Combine.perKey、Flatten合并多个 PCollection、Partition拆分Windowing固定窗口、滑动窗口、会话窗口、全局窗口默认Triggers控制结果何时发出典型组合如AfterWatermark.pastEndOfWindow().withEarlyFirings(...)配合withAllowedLateness(...)Side InputsParDo 的附加输入如View.asMap()做查表Schema强类型访问结构化数据如DefaultSchema(AutoValueSchema.class)Error Handling通过TupleTag实现死信队列Dead Letter Queue模式。该技能还特别强调/model目录是 Beam 模型与可移植框架的语言无关规范Protocol Buffer 与 gRPC由pipelineRunner API、fn-executionFn API、job-managementJob/Expansion/Artifact API、interactive交互式 API四部分组成URN 字符串如beam:transform:pardo:v1是跨语言变换的 API 契约修改 proto 时必须保持向后兼容并重新生成语言绑定./gradlew :model:pipeline:generateProto。5.4 I/O 连接器使用与开发io-connectors / developing-new-io-connectorsBeam 在sdks/java/io/下提供 51 个 Java I/O 连接器覆盖云存储BigQuery、Bigtable、Spanner、Pub/Sub、GCS、数据库JDBC、MongoDB、Cassandra、HBase、Redis、Neo4j、ClickHouse、Elasticsearch、消息队列Kafka、Pulsar、RabbitMQ、AMQP、JMS、MQTT、Solace、文件格式Parquet、CSV、JSON、XML、Thrift、Iceberg等类别。集成测试框架位于it/目录it/common、it/google-cloud-platform/、it/kafka/、it/testcontainers/等。现代 Beam 连接器的开发标准见 developing-new-io-connectors要求schema-aware支持readRows()输出 Beam Row、可通过 Expansion Service 跨语言使用Python/Go 流水线、无缝集成 Beam YAML 与 Managed I/O API。开发步骤包括在sdks/java/io/connector-name/下建模块build.gradle使用org.apache.beam.module插件采用 AutoValue builder 模式实现核心IO变换读侧优先用 Splittable DoFnSDF替代遗留 Source 类以获得动态再平衡与 watermark 支持在model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/external_transforms.proto中为读写变换定义 URN实现SchemaTransformProvider如MyReadSchemaTransformProvider extends TypedSchemaTransformProviderConfiguration暴露给跨语言流水线与 YAML在sdks/java/managed/src/main/java/org/apache/beam/sdk/managed/Managed.java中注册READ_TRANSFORMS/WRITE_TRANSFORMS将模块加入sdks/java/io/expansion-service/build.gradle的运行时依赖在sdks/python/apache_beam/transforms/managed.py与external.py中完成 Python 侧注册分层验证单元测试:sdks:java:io:connector-name:test、Managed 翻译测试:sdks:java:managed:test、集成测试依赖it/下的ResourceManager。写侧Sink开发有四种推荐模式基于 DoFn/批处理如 Kafka、Pub/Sub、NoSQL 的GroupIntoBatches、两阶段提交实现精确一次关系型数据库、Iceberg、Delta Lake、基于FileIO的文件写入、以及通过PCollectionRowTuple输出失败记录的死信报告。需要特别警惕 runner 对变换的失败重试可能造成目标系统数据重复。5.5 DoFn 现代化改造beam-dofn-modernizerBeam 已转向在DoFn方法中使用参数注入parameter injection以提升可读性与执行效率。技能给出完整的映射表遗留 Context 用法如ProcessContext c现代参数替代c.element()Element T elementc.timestamp()Timestamp Instant timestampc.pane()PaneInfo panec.window()BoundedWindow windowc.sideInput(PCollectionViewT view)SideInput(viewName) T valuec.getPipelineOptions()PipelineOptions optionsc.output(value)OutputReceiverT receiver后receiver.output(value)c.output(tag, value)MultiOutputReceiver receiver后receiver.get(tag).output(value)c.outputWithTimestamp(value, ts)OutputReceiverT receiver后receiver.outputWithTimestamp(value, ts)例如遗留写法ProcessElement public void processElement(ProcessContext c) { c.output(transform(c.element())); }应改写为ProcessElement public void processElement(Element T element, OutputReceiverV receiver) { receiver.output(transform(element)); }。Side input 场景则需同时改动流水线侧ParDo.of(new MyFn()).withSideInput(sideInputName, myView)与 DoFn 侧SideInput(sideInputName) String value。若某个 side input 可选且 DoFn 存在条件逻辑最佳实践是拆分为两个 DoFn 类以保证类型安全。5.6 元数据扩展与 CI/CD、贡献流程adding-new-metadata / ci-cd / contributingadding-new-metadata 是一份贯穿执行引擎全链路的深水区指南为WindowedValue新增元数据如 CDC 元数据、drain 模式标志、OpenTelemetry trace context时需依次修改model/fn-execution/.../beam_fn_api.protoElementMetadata消息、sdks/java/core/src/main/java/org/apache/beam/sdk/values/WindowedValue.java接口及其全部实现ValueInGlobalWindow、ValueInSingleWindow、ValueInEmptyWindows、OutputBuilder采用 builder 模式而非给outputWindowedValue(...)加参数、Dataflow Windmill 的WindmillSink/UngroupedWindmillReader/WindowingWindmillReader序列化链路以及SimpleDoFnRunner、StatefulDoFnRunner、ReduceFnRunner等传播路径若需向用户暴露还要更新DoFnSignatures与ByteBuddyDoFnInvokerFactory。文档末尾附有 12 项检查清单任何遗漏都会导致元数据在执行中被静默丢弃。ci-cd 说明 Beam 使用 GitHub Actions 做 CI/CD工作流分为 PreCommitPR 与合并时运行命名beam_PreCommit_*.yml、PostCommit合并后与定时运行命名beam_PostCommit_*.yml、Scheduled每晚检查外部依赖影响。涉及 GCP 的工作流依赖GCP_PROJECT_ID、GCP_REGION、GCP_TESTING_BUCKET、GCP_SA_EMAIL、GCP_SA_KEY等 secrets。contributing 描述了完整的贡献流程通过 issue 认领.take-issue、为大规模改动先在 devbeam.apache.org 邮件列表讨论并撰写设计文档、所有源文件必须带 Apache 许可头、新依赖须为 Apache 兼容的开源许可。特别值得注意的是一条AGENT DIRECTIVE修改核心组件sdks/java/core、runners/core-*、model、sdks/python/apache_beam/transforms|coders、sdks/go/pkg/beam/core、FileIO、GCS等见.github/autolabeler.yml的core标签风险极高容易在分布式 runner 上引入静默回归Agent 需要以更严苛的标准评估改动。本地预提交验证命令为./gradlew javaPreCommit、./gradlew :sdks:python:test、./gradlew :sdks:go:test。六、如何新增一个技能遵循 README 中给出的三步流程即可向仓库添加新技能在.agent/skills/下创建新的技能文件夹在其中添加SKILL.md文件头写入 YAML frontmatter必须包含name与description字段在 Markdown 正文中编写清晰、可执行的指令。结合仓库现有技能的最佳实践新增时还应注意几点description应同时概括技能内容并说明适用时机便于 Agent 在 Discovery 阶段命中正文指令要具体到可复现的命令、代码与文件路径参考gradle-build对./gradlew命令、beam-concepts对 Java/Python 双语言代码示例的写法文件头部保留 Apache License 2.0 许可头如技能涉及多个领域的交叉引用如io-connectors指向developing-new-io-connectors应使用仓库根目录相对路径进行链接。七、总结.agent/skills/目录是 Apache Beam 面向 AI Agent 时代的基础设施它用统一的SKILL.md YAML frontmatter 结构把 Beam 数十年积累的领域知识——从核心编程模型、三大 SDK 的开发实践、51 I/O 连接器到构建系统、CI/CD、核心执行引擎的深层改造——编码为 Agent 可发现、可激活、可执行的操作指南。这套技能体系既服务于日常的构建测试与贡献流程也覆盖了WindowedValue元数据扩展、DoFn 现代化等高难度改造场景是 Agent 深入理解并高效操刀 Apache Beam 代码库的首选入口。对于希望让 AI 编程助手开箱即用地熟悉 Beam 的开发者与贡献者而言直接浏览上述各技能的SKILL.md是最快的上手路径。赞分享大数据批处理流处理数据工程【免费下载链接】beamApache Beam is a unified programming model for Batch and Streaming data processing.项目地址https://gitcode.com/gh_mirrors/beam4/beam点击查看免费下载相关推荐PyPTO Agent Skills 专家技能集面向框架调试、Pass 分析与 PR 流程的 AI 工程化体系PyPTO Agent Skills 专家技能集面向框架调试、Pass 分析与 PR 流程的 AI 工程化体系 本文系统介绍 CANN PyPTO 仓库内置的人工智能编译器模型编译深度学习高性能计算CANNAscendJulia 仓库的 Agent Skills面向 AI 开发者的项目内技能体系详解Julia 仓库的 Agent Skills面向 AI 开发者的项目内技能体系详解 导读 本文介绍 Julia 语言官方仓库JuliaLang/julia编程语言编译器语言运行时标准库JIT编译Flutter 仓库 Android Agent面向 Android 团队的专用 AI 代理配置与 Skills 机制解析Flutter 仓库 Android Agent面向 Android 团队的专用 AI 代理配置与 Skills 机制解析 Flutter 主仓库的 .age跨平台移动开发前端UI组件桌面应用上一篇TF Serving集成TF Adapter下一篇从论文到代码Pointer Summarizer如何实现Get To The Point论文精髓创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考