批处理流处理大数据【免费下载链接】beamApache Beam is a unified programming model for Batch and Streaming data processing.项目地址https://gitcode.com/gh_mirrors/beam15/beam点击查看免费下载Apache Beam Python SDK 内置的 DataFrame API 是一套基于 pandas 实现的领域特定语言DSL它允许开发者在 Beam 管道中以熟悉的 pandas 方法操作大规模数据集同时享受 Beam 的分布式并行处理与批流统一编程模型。本文以仓库中的官方概述文档 website/www/site/content/en/documentation/dsls/dataframes/overview.md 为主线结合 sdks/python/apache_beam/dataframe 目录下的源码实现系统讲解 DataFrame API 的环境准备、核心用法、与普通 PCollection 管道的互操作方式以及其与原生 pandas 的差异与限制。读完本文你将能够使用read_csv、groupby、to_csv等 pandas 惯用写法构建完整的分布式数据处理管道并掌握to_dataframe/to_pcollection/DataframeTransform三种集成模式。什么是 Beam DataFramesApache Beam Python SDK 提供了一个用于处理 pandas 风格 DataFrame 对象的 API。它的核心思想是你可以把一个 PCollection 转换为 DataFrame然后使用 pandas DataFrame API 的标准方法与之交互。从实现上看DataFrame API 构建在 pandas 实现之上pandas 的 DataFrame 方法会被并行地调用在数据集的各个子集上参见 sdks/python/apache_beam/dataframe/transforms.py 中关于分区目标大小TARGET_PARTITION_SIZE 1 238MB、MIN_PARTITION_SIZE 1 190.5MB以及DEFAULT_PARTITIONS 100等分区参数的配置。Beam DataFrames 与 pandas DataFrames 最大的区别在于所有操作都被 Beam API 延迟deferred执行以适配 Beam 的并行处理模型。也就是说你在本地写下的 DataFrame 操作并不会立即计算而是被构建成一棵表达式树最终由 Beam 运行时转化为实际的 PTransform 计算图去执行。你可以把 Beam DataFrames 视为面向 Beam 管道的领域特定语言DSL。与 Beam SQL仓库中的 website/www/site/content/en/documentation/dsls/sql/overview.md类似DataFrames 是内置于 Beam Python SDK 的 DSL。使用这套 DSL你可以不用显式引用 ParDo、CombinePerKey 等标准 Beam 构造对应文档见 website/www/site/content/en/documentation/transforms/python/elementwise/pardo.md 与 website/www/site/content/en/documentation/transforms/python/aggregation/combineperkey.md从而用更熟悉的编程界面完成管道编写。Beam DataFrame API 的定位是在 Beam 管道内提供熟悉的编程接口。在某些场景下DataFrame API 还能通过把操作委托给高度优化、向量化的 pandas 实现来提升管道效率——这正是 DataFrame DSL 相对手写逐元素 ParDo 的潜在优势。前置条件安装与版本要求要使用 Beam DataFrames需要满足以下环境条件Beam Python 版本 2.26.0 或更高完整安装说明见 website/www/site/content/en/get-started/quickstart-py.md 的 Python SDK 快速入门受支持的 pandas 版本。在 Beam 2.34.0 及更新版本中最简单的方式是使用dataframe安装扩展pip install apache_beam[dataframe]该扩展在仓库 sdks/python/setup.py 中定义的依赖约束为dataframe_dependency [ pandas1.4.3,!1.5.0,!1.5.1,2.3;python_version3.8, ]即要求 pandas 版本不低于 1.4.3、排除 1.5.0 与 1.5.1因 pandas 上游问题 pandas-dev/pandas#45725且上限低于 2.3同时要求 Python 3.8 及以上。分布式执行时的重要注意事项在分布式 Runner 上运行 DataFrame API 管道时worker 上必须安装与本地相同的 pandas 版本。可以参考仓库 sdks/python/container 目录下各 Python 版本的base_image_requirements.txt例如 sdks/python/container/py38/base_image_requirements.txt、sdks/python/container/py39/base_image_requirements.txt来确认对应 Python 版本和 Beam 发行版在 worker 上默认使用的 pandas 版本如 py38 镜像固定pandas2.0.3、py39 镜像固定pandas2.1.4从而保证本地与远端行为一致。使用 DataFrames完整的出租车数据聚合示例官方概述文档给出了一个非常直观的示例读取纽约市出租车数据 CSV 文件执行分组聚合再把结果写回 CSV。仓库中的完整实现位于 sdks/python/apache_beam/examples/dataframe/taxiride.py其核心聚合逻辑为from apache_beam.dataframe.io import read_csv with pipeline as p: rides p | read_csv(input_path) # 按 DOLocationID 分组统计每个下车地点的乘客总数 agg rides.groupby(DOLocationID).passenger_count.sum() agg.to_csv(output_path)pandas 能够从 CSV 数据的首行推断出列名这正是passenger_count和DOLocationID这两个列名的来源。在run_aggregation_pipeline之外sdks/python/apache_beam/examples/dataframe/taxiride.py 还提供了run_enrich_pipeline这一增强版示例将出租车行程数据与区域查找表通过merge关联再按行政区Borough聚合乘客数with pipeline as p: rides p | Read taxi rides read_csv(input_path) zones p | Read zone lookup read_csv(zone_lookup_path) # 先让 zones.LocationID 成为索引再与 rides.DOLocationID 做左连接 rides rides.merge( zones.set_index(LocationID).Borough, right_indexTrue, left_onDOLocationID, howleft) # 按 Borough 汇总乘客数 agg rides.groupby(Borough).passenger_count.sum() agg.to_csv(output_path)该示例展示了 DataFrame API 在典型批处理分析场景读 CSV → 关联 → 分组聚合 → 写 CSV中的完整用法。独立于传统 Beam 原语的 IO 能力在上面的例子中唯一的传统 Beam 类型是Pipeline实例其余部分完全用 DataFrame API 编写。这之所以可行是因为 Beam DataFrame API内置了自己的 IO 操作例如read_csv和to_csv它们基于 pandas 原生实现源码见 sdks/python/apache_beam/dataframe/io.py。read_*与to_*操作的特点包括支持文件模式file patterns以及任何 Beam 兼容的文件系统GCS、S3、本地文件等read_csv以增量incrementalTrue方式分块读取数据默认按约 10000 行_DEFAULT_LINES_CHUNKSIZE切块见 sdks/python/apache_beam/dataframe/io.py对于大文件且记录中不包含引号内换行的情况可以为read_csv传入splittableTrue启用基于换行的动态拆分dynamic splitting从而获得更好的并行度但若记录中包含被引号包裹的换行使用该选项可能导致记录被截断、数据损坏在 DataFrame 之外模块还提供了read_gbq读取 BigQuery 表支持通过use_bqstorage_api切换DIRECT_READ/EXPORT两种读取方式、read_fwf、read_json、read_html以及read_excel/read_feather/read_parquet/read_stata等二进制格式读取见 sdks/python/apache_beam/dataframe/io.py。需要说明的是read_csv与read_json目前不支持nrows参数源码中会显式抛出异常或NotImplementedError。分组操作通过 group-by-key 完成而在最终to_csv写入之前可以应用任意 pandas 操作本例中是sum。从源码实现看to_csv会先把 DataFrame 通过convert.to_pcollection(df, yield_elementspandas)转成包含 DataFrame 元素的 PCollection再交给_WriteToPandas经由fileio.WriteToFiles写出sdks/python/apache_beam/dataframe/io.py。运行方式上述taxiride.py支持通过命令行参数选择运行两种管道见 sdks/python/apache_beam/examples/dataframe/taxiride.py--pipeline location_id_agg默认按DOLocationID聚合--pipeline borough_enrich关联区域表后按Borough聚合--input输入 CSV 路径默认gs://apache-beam-samples/nyc_taxi/misc/sample.csv--output输出路径必填--zone_lookup区域查找表 CSV 路径。例如python -m apache_beam.examples.dataframe.taxiride \ --pipeline location_id_agg \ --input gs://apache-beam-samples/nyc_taxi/misc/sample.csv \ --output ./output_agg.csvBeam DataFrame API 的目标是与 pandas 原生实现保持兼容但存在少量需要注意的差异详见仓库文档 website/www/site/content/en/documentation/dsls/dataframes/differences-from-pandas.md。将 DataFrames 嵌入管道在更大的管道中你可以把 DataFrame API 与普通 PCollection 结合起来把 PCollection 转成 DataFrame处理完后再转回 PCollection。为此参与转换的 PCollection 必须带有 schema模式——这种 PCollection 被称为schema-aware PCollection。关于 schema 的定义参见 website/www/site/content/en/documentation/programming-guide.md第 6 章 Schemas6.1 What is a schema?创建 schema 的方法参见 website/www/site/content/en/documentation/programming-guide.md6.5 Creating Schemas。仓库中的完整 wordcount 示例位于 sdks/python/apache_beam/examples/dataframe/wordcount.py其核心流程如下from apache_beam.dataframe.convert import to_dataframe from apache_beam.dataframe.convert import to_pcollection with beam.Pipeline(optionsPipelineOptions(pipeline_args)) as p: lines p | Read ReadFromText(known_args.input) words ( lines | Split beam.FlatMap( lambda line: re.findall(r[\w], line)).with_output_types(str) # 映射为 Row 对象以生成适合转换为 dataframe 的 schema | ToRows beam.Map(lambda word: beam.Row(wordword))) df to_dataframe(words) df[count] 1 counted df.groupby(word).sum() counted.to_csv(known_args.output) # Deferred DataFrame 也可以转换回带 schema 的 PCollection counted_pc to_pcollection(counted, include_indexesTrue) _ ( counted_pc | beam.Filter(lambda row: row.count 50) | beam.Map(lambda row: f{row.word}: {row.count}) | beam.Map(print))这个示例清晰展示了三种风格的融合用beam.FlatMap拆分文本、用beam.Map(lambda word: beam.Row(wordword))把普通元素包装为beam.Row从而让 PCollection 具备 schema用to_dataframe(words)把 schema-aware PCollection 转成 Deferred DataFrame用df.groupby(word).sum()完成 pandas 风格的分组计数用to_csv写出结果同时用to_pcollection(counted, include_indexesTrue)把 DataFrame 转回 PCollection继续用beam.Filter、beam.Map等传统变换做后续处理。to_dataframe 与 to_pcollection 的底层行为从源码 sdks/python/apache_beam/dataframe/convert.py 看to_dataframe(pcoll, proxyNone, labelNone)的工作方式是若未提供proxy则要求输入 PCollection必须已定义 schema即pcoll.element_type非空否则抛出ValueError提示“请为输入 PCollection 指定 schema 类型或提供 proxy”根据 proxy 的类型DataFrame 或 Series选择RowsToDataFrameFn或ElementsToSeriesFn对 PCollection 做批量化batching再包装为一个PlaceholderExpression返回DeferredFrame。to_pcollection(*dataframes, labelNone, always_return_tupleFalse, yield_elementsschemas, include_indexesFalse, pipelineNone)sdks/python/apache_beam/dataframe/convert.py则会真正创建并应用 Beam 操作来计算这些 deferred DataFrame返回其结果 PCollectionyield_elements可选schemas默认逐行展开、DataFrame 转为 schema-aware PCollection列值可通过属性访问或pandas直接产出原始 pandas 对象include_indexesTrue时会把 DataFrame 的索引列尝试纳入输出 schema要求索引层级均已命名且名字与列名不冲突它同样接受非 deferred 的普通 pandas DataFrame——此时整份 DataFrame 会被序列化进计算图因此数据量很大时更推荐先写盘再用read_*读入多个相关结果可以一次性传入to_pcollection同时计算这通常比分别转换更高效。此外to_pcollection内部通过_DataframeExpressionsTransform见 sdks/python/apache_beam/dataframe/transforms.py把表达式树划分成一系列可融合的“stage”每个 stage 会根据操作所需的分区要求partitionings.Index/Singleton/JoinIndex/Arbitrary定义见 sdks/python/apache_beam/dataframe/partitionings.py决定是否 shuffle并用beam.CoGroupByKey把数据重新分区、用_PreBatch/_ReBatch调整批次大小后在每个分区上调用 pandas 方法完成实际计算。这正是“pandas 方法被并行地调用在数据子集上”这一设计的具体落地。使用 DataframeTransform 传递函数除了手动to_dataframe/to_pcollection转换还可以把函数直接传给DataframeTransformfrom apache_beam.dataframe.transforms import DataframeTransform with beam.Pipeline() as p: ... | beam.Select(DOLocationIDlambda line: int(..), passenger_countlambda line: int(..)) | DataframeTransform(lambda df: df.groupby(DOLocationID).sum()) | beam.Map(lambda row: f{row.DOLocationID},{row.passenger_count}) ...DataframeTransform与 Beam SQL DSL 中的SqlTransform类似SqlTransform把一条 SQL 查询翻译为一个 PTransform而DataframeTransform是一个接收并返回 DataFrame 的函数的 PTransform。当某个独立函数既可以运行在 Beam 上、又可以运行在普通 pandas DataFrame 上时DataframeTransform尤其有用——同一份业务逻辑代码可以在本地调试与分布式执行之间无缝复用。DataframeTransform 多输入多输出DataframeTransform可以通过按名与按关键字接收和返回多个 PCollection官方文档给出如下几种形态# 元组输入多个 PCollection 作为位置参数传给函数 output (pc1, pc2) | DataframeTransform(lambda df1, df2: ...) # 字典输入PCollection 作为关键字参数传给函数 output {a: pc, ...} | DataframeTransform(lambda a, ...: ...) # 元组输出函数返回多个 DataFrame解包为多个 PCollection pc1, pc2 {a: pc} | DataframeTransform(lambda a: expr1, expr2) # 字典输出函数返回字典得到多个命名的 PCollection {...} {a: pc} | DataframeTransform(lambda a: {...})从实现看sdks/python/apache_beam/dataframe/transforms.pyDataframeTransform.expand会先把输入 PCollection 扁平化为字典逐个调用convert.to_dataframe得到输入 DataFrame调用用户函数得到结果 DataFrame 后再通过convert.to_pcollection(..., always_return_tupleTrue)统一求值最后按原有嵌套结构还原返回值。DataframeTransform 的可选参数DataframeTransform(func, proxyNone, yield_elementsschemas, include_indexesFalse)的完整参数见 sdks/python/apache_beam/dataframe/transforms.py 的 docstringfunc接收并返回 DataFrame或 Series的函数yield_elements默认为schemas即输出逐行的 schema-aware PCollection每行一个元素、列可通过属性访问设为pandas则直接输出原始 pandas 对象include_indexes默认为False当yield_elementsschemas时若为True则尝试把 DataFrame 索引列纳入输出 schema要求索引层级均有名字且名字不与列名重复否则会报错proxy当输入 PCollection 的元素本身就是 DataFrame/Series而非带 schema 的行时需要提供一个空的、dtype 与元素一致的 pandas DataFrame/Series 实例作为类型原型若输入 PCollection 已带 schema 则该参数被忽略。与 pandas 的差异你需要知道的行为边界Beam DataFrame API 的目标是成为 pandas 的“drop-in replacement”可直接替换的实现但存在若干差异。仓库文档 website/www/site/content/en/documentation/dsls/dataframes/differences-from-pandas.md 系统梳理了这些差异底层错误类型定义在 sdks/python/apache_beam/dataframe/frame_base.py 与 sdks/python/apache_beam/dataframe/expressions.py。读取 pandas 源数据的差异Beam 操作总是与某个 Pipeline 绑定。要把源数据读入 Beam DataFrame必须把 source 应用|到 pipeline 对象上df p | beam.dataframe.io.read_csv(...)这与 pandas 的read_csv类似但这里的df是一个延迟的 Beam DataFrame代表文件内容。输入文件名可以是任意被fileio.MatchFiles理解的文件模式。五类不支持或受限支持的操作1. 不可并行化的操作NonParallelOperation典型例子DeferredDataFrame.quantile、DeferredDataFrame.mode。为了支持分布式处理Beam 会把 DataFrame 操作并行地调用在数据子集上。某些操作无法并行化默认会抛出NonParallelOperation错误。从源码看sdks/python/apache_beam/dataframe/expressions.py当某个ComputedExpression要求Singleton分区即所有数据必须集中到一个分区而当前未开启允许非并行操作时就会抛出该异常。变通方法可以用beam.dataframe.allow_non_parallel_operations上下文管理器包住这类操作from apache_beam import dataframe with dataframe.allow_non_parallel_operations(): quantiles df.quantile()注意这会把整个输入数据集收集到单个节点上执行存在内存溢出风险只应在确认输入足够小、能在单 worker 上处理时使用。对应的上下文管理器实现见 sdks/python/apache_beam/dataframe/expressions.py。2. 产生非延迟列的操作WontImplementError典型例子DeferredDataFrame.pivot、DeferredDataFrame.transpose、DeferredSeries.factorize。Beam DataFrame 操作是延迟的但结果 DataFrame 的 schema 不是延迟的——结果列必须在没有数据的情况下就能计算出来。某些操作无法满足这一要求因而无法实现调用时会抛出WontImplementError。目前没有现成的变通方案未来可能通过 categorical 列支持非延迟列操作。3. 产生非延迟值或绘图的操作WontImplementError典型例子DeferredSeries.to_list、DeferredSeries.array、DeferredDataFrame.plot。由于 Beam 是延迟 API产生非延迟值或绘图的操作不可能实现调用即抛出WontImplementError。变通方法如果使用 Interactive Beam可以用collect把数据集拉到本地内存后再执行这些操作。4. 顺序敏感的操作WontImplementError典型例子DeferredDataFrame.head、DeferredSeries.diff、DeferredDataFrame.interpolate。Beam PCollection 本质上是无序的因此对行顺序敏感的操作不受支持调用时抛出WontImplementError。变通方法一是借助 Interactive Beam 的collect拉取到本地二是改写为不敏感于顺序的代码——例如 pandas 用户常用head来“瞄一眼”数据如果只是想查看元素子集可以用不需要先收集数据的sample类似地可以用nlargest代替sort_values(...).。5. 产生延迟标量的操作有些 DataFrame 操作会产生延迟标量deferred scalar。在 Beam 中实际计算被延迟因此这些值不能用于控制流。例如可以计算Series.sum但不能立即对其结果做分支判断Series.is_unique也是类似的例子。对延迟标量做真值测试如用于if分支会抛出TypeError对应实现见 sdks/python/apache_beam/dataframe/frame_base.py 中_DeferredScalar.__bool__的明确拒绝。尚未实现的操作Beam DataFrame API 已实现大量常用的 pandas 操作且社区仍在持续推进剩余操作的支持但 pandas API 体量庞大仍存在缺口。如果调用尚未实现的操作会抛出NotImplementedError。借助 Interactive Beam 使用完整 pandas APIInteractive Beam惯例导入为ib是专为交互式 notebook 设计的模块它提供的ib.collect函数可以把 PCollection 或 deferred DataFrame 拉取到本地内存变成一个普通 pandas DataFrame。materialize 之后你就可以执行 pandas API 中的任意操作而不仅仅局限于 Beam 支持的那些。快速入门可参考 website/www/site/content/en/get-started/_index.md 中的 Try Apache Beam 体验入口。小结与推荐阅读路径Beam DataFrame API 为 Beam 管道提供了一条“pandas 语法、Beam 引擎”的写作路径通过内置的read_*/to_*IO、to_dataframe/to_pcollection转换工具和DataframeTransformPTransform开发者可以在批流统一的分布式引擎上复用熟悉的 pandas 编程心智同时获得向量化 pandas 实现带来的性能收益而其与 pandas 的行为差异非并行、非延迟列、顺序敏感、延迟标量等则由明确的异常类型在构建期即暴露出来。如需进一步深入建议按以下路径在仓库中继续阅读官方概述文档website/www/site/content/en/documentation/dsls/dataframes/overview.md差异与限制详解website/www/site/content/en/documentation/dsls/dataframes/differences-from-pandas.md完整示例sdks/python/apache_beam/examples/dataframe/taxiride.py出租车聚合/关联、sdks/python/apache_beam/examples/dataframe/wordcount.py与 PCollection 互操作核心源码sdks/python/apache_beam/dataframe/convert.pyto_dataframe/to_pcollection、sdks/python/apache_beam/dataframe/io.pyread_*/to_*、sdks/python/apache_beam/dataframe/transforms.pyDataframeTransform与表达式求值 stage、sdks/python/apache_beam/dataframe/frames.pyDeferredDataFrame/DeferredSeries实现测试与验证sdks/python/apache_beam/dataframe/convert_test.py、sdks/python/apache_beam/dataframe/frames_test.py、sdks/python/apache_beam/dataframe/io_test.py、sdks/python/apache_beam/dataframe/transforms_test.py 可帮助你理解各类操作的行为边界。赞分享批处理流处理大数据【免费下载链接】beamApache Beam is a unified programming model for Batch and Streaming data processing.项目地址https://gitcode.com/gh_mirrors/beam15/beam点击查看免费下载相关推荐Apache Beam Python SDK DataFrame API 预览指南用 pandas 风格代码编写分布式批处理与流处理管道Apache Beam Python SDK DataFrame API 预览指南用 pandas 风格代码编写分布式批处理与流处理管道 本文基于 dataf大数据批处理流处理数据工程Apache Beam Python DataFrame API 完全指南在 Beam 管道中使用 pandas 风格 DSLApache Beam Python DataFrame API 完全指南在 Beam 管道中使用 pandas 风格 DSL Apache Beam Pyt大数据批处理流处理数据工程Apache Beam YAML API 完全指南用声明式 YAML 编写批处理与流处理管道Apache Beam YAML API 完全指南用声明式 YAML 编写批处理与流处理管道 Apache Beam YAML API 是 Beam Pyth上一篇WaveTools免费开源的游戏优化工具提升《鸣潮》体验的终极解决方案下一篇如何零基础入门扩散模型Awesome-Diffusion-Models 精选12篇顶级教程博客完整指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考