你好我是程序员贵哥。今天我要与你分享的主题是“为什么Beam要如此抽象封装数据”。很多人在刚开始接触Apache Beam的时候都会觉得这里面的概念太抽象了。什么PCollection、PValue、Transform……这都是些什么尤其是PCollection完全和先前的技术知识找不到对应。确实如此。同样作为数据的容器PCollection却并不像Python/Java的List或者C的vector。PCollection是无序的Beam对于PCollection中元素的处理顺序不作任何保证。所以你不可能说“我想处理PCollection中的第二个元素”因为它就没有“第几个”这种概念。PCollection也不像Python/Java的Set或者C的unordered_setPCollection不一定有固定的边界。所以你也不能指望去查找一个PCollection的大小。在PCollection的世界里也没有“固定大小”这样的概念。作为程序员我很讨厌重复造轮子尤其是新瓶装旧酒。的确有很多开发者为了体现自己项目的复杂度故意强行引进了很多概念让大家都似懂非懂的。这就像是为了体现自己知道茴香豆的“茴”有几种写法一样故意用另一种写法来体现自己“有文化”。那么Apache Beam引进这么多概念是不是也是故意强行造轮子呢答案是否定的。这一讲我们就要分析PCollection为什么会如此设计。在之前的章节中我们已经讲到了Apache Beam的“爷爷”——FlumeJava产生的背景。当时Google的工程师们发现MapReduce使用起来并没有很方便。如果计算过程都能够分解成map、shuffle、reduce步骤的话MapReduce是非常能胜任的。但是很多现实世界中的问题往往需要一系列的map或reduce步骤。这样的数据流水线就需要额外的代码用来协调这些map或reduce步骤。也需要一些额外的代码去管理分步的map/reduce步骤产生的一些中间结果。以至于新的开发者很难理解复杂的流水线。清楚这样的背景对于理解Apache Beam的PCollection起源很有帮助。因为这个项目起源只是为了提供比MapReduce更好的开发体验最终的运算引擎仍然是MapReduce。为什么需要PCollection那么为什么Beam需要PCollection这样一个全新的抽象数据结构呢我们知道不同的技术系统有不同的数据结构。比如C里有vector、unordered_map安卓有ListView。相比它们而言其实Beam的数据结构体系是很单调的几乎所有数据都能表达为PCollection。PCollection就是Parallel Collection意思是可并行计算的数据集。如果你之前学习了Spark的章节就会发现PCollection和RDD十分相似。在一个分布式计算系统中我们作为架构设计者需要为用户隐藏的实现细节有很多其中就包括了数据是怎样表达和存储的。这个数据可能是来自于内存的数据内部可能只是由一个C array存储也有可能是来自外部文件由几个文件存储或者是来自于MySQL数据库由数据库的行来表达。如果没有一个统一的数据抽象的话开发者就需要不停地更改代码。比如在小规模测试的时候用C vector等到了真实的生产环境我再换MySQL row。沉溺于这样的实现细节会让开发者无法集中注意力在真正重要的事情上那就是“你想怎样处理数据”。清楚了这些你就能明白我们需要一层抽象来表达数据而这层抽象就是PCollection。PCollection的创建完全取决于你的需求。比如在测试中PCollection往往来自于代码生成的伪造数据或者从文件中读取。Pythonlines (p | beam.Create([To be, or not to be: that is the question. ])) lines p | ReadMyFile beam.io.ReadFromText(gs://some/inputData.txtJavaPCollectionString lines p.apply( ReadMyFile, TextIO.read().from(protocol://path/to/some/inputData.txt));为什么PCollection需要Coders与普通编程相比PCollection的另一个不同之处是你需要为PCollection的元素编写Coder。例如你有一个自己的类MyClass那么PCollection一定需要实现Coder。刚开始使用Beam时你可能会感觉这很不方便。例如你只是要去读取MySQL的一个表也得为此实现Coder。Coder的作用和Beam的本质紧密相关。因为你的计算流程最终会运行在一个分布式系统。所以所有的数据都有可能在网络上的计算机之间相互传递。而Coder就是在告诉Beam怎样把你的数据类型序列化和逆序列化以方便在网络上传输。Coder需要注册进全局的CoderRegistry中简单来说是为你自己的数据类型建立与Coder的对应关系。不然每次你都需要手动指定Coder。Pythonapache_beam.coders.registry.register_coder(int, BigEndianIntegerCoder)JavaPipelineOptions options PipelineOptionsFactory.create(); Pipeline p Pipeline.create(options); CoderRegistry cr p.getCoderRegistry(); cr.registerCoder(Integer.class, BigEndianIntegerCoder.class);为什么PCollection是无序的讲完为什么PCollection需要Coder之后我们再来看下为什么PCollection是无序的。PCollection的无序特性其实也和它的分布式本质有关。一旦一个PCollection被分配到不同的机器上执行那么为了保证最大的处理输出不同机器都是独立运行的。所以它的执行顺序就无从得知了。可能是第一个元素先被运行也可能是第二个元素先被运行。所以肯定不会有PCollection[2]这样的运算符。为什么PCollection没有固定大小无序也就算了为什么PCollection还没有固定大小呢前面的章节中讲到过Beam想要统一批处理和流处理所以它要统一表达有界数据和无界数据。正因为如此PCollection并没有限制它的容量。如前面所说它可能表达内存上的一个数组也可能表达整个数据库的所有数据。一个PCollection可以是有界的也可以是无界的。一个有界的PCollection表达了一个已知大小的固定的数据集。一个无界的PCollection表达了一个无限大小的数据集。事实上一个PCollection是否有界往往取决于它是如何产生的。从批处理的数据源中读取比如一个文件或者是一个数据库就会产生有界的PColleciton。如果从流式的或者是持续更新的数据库中读取比如pub/sub或者kafka会产生一个无界的PCollection。但是PCollection的有界和无界特性会影响到Beam的数据处理方式。一个批处理作业往往处理有界数据。而无界的PCollection需要流式的作业来连续处理。在实现中Beam也是用window来分割持续更新的无界数据。所以一个流数据可以被持续地拆分成不同的小块。这样的处理方式我们会在实战部分展开。如何理解PCollection的不可变性在普通编程语言中大部分数据结构都是可变的。PythonAlist [] alist.append(1)CStd::vectorint list; list.push_back(1);但是PCollection不提供任何修改它所承载数据的方式。修改一个PCollection的唯一方式就是去转化(Transform)它下一讲会展开讲Transformation。但是在这一讲我们需要理解的是Beam的PCollection都是延迟执行deferred execution的模式。也就是说当你下面这样的语句的时候什么也不会发生。JavaPCollectionT1 p1 ...; PCollectionT2 p2 doSomeWork(p1);这样的语句执行完p2这个PCollection仅仅会记录下自己是由doSomeWork这个操作计算而来的和计算自己所需要的数据p1。当你执行写完100行的beam的运算操作最终的结果仅仅是生成了一个有向无环图DAG也就是执行计划execution plan。为什么这么设计呢如果你记得我们在专栏第一部分讲到的大规模数据框架设计可能会有印象。这样的有向无环图是框架能够自动优化执行计划的核心。类似图中这样的数据处理流程在Beam获知了整个数据处理流程后就会被优化成下图所示。这样的优化在Beam中被称为sibling fusion。类似的操作优化我们后面会继续介绍。在这个小标题下我想用这个优化的例子说明PCollection下的数据不可变是因为改变本身毫无意义。例如在刚才这个例子中你会发现优化后的执行计划里已经没有了数据A0。因为Beam发现数据A0作为中间结果并不影响最终输出。另外由于Beam的分布式本质即使你想要去修改一个PCollection的底层表达数据也需要在多个机器上查找毫无实现的价值。小结这一讲我们介绍了整个Beam核心数据结构PCollection的设计思想。尤其是分析了PCollection的几大特性为什么是这么设计的。它的特性包括Coders、无序性、没有固定大小、不可变性等。在每个特性下面我们也介绍了几个例子或者代码示例希望能帮助你加深印象。