这是关于Yelp的实时流数据根底设施系列文章的第三篇。这个系列会深度解说咱们怎样用“确保只需一次”的办法把MySQL数据库中的改动实时地以流的办法传输出去，咱们怎样主动盯梢表形式改动，怎样处理和转化流，以及终究怎样把这些数据存储到Redshift或Salesforce之类的数据仓库中去。 当你的体系每天要实时从MySQL到Kafka发布几十亿条音讯时，你会怎样办理这些数据的形式信息呢?当你的体系要接入几百个服务时，你就要处理几千种不同的形式，手艺办理是不可行的。有必要有主动化的方案来处理从上游数据源到一切下流顾客的形式改动问题。Confluent公司的Schema Registry和Kafka Connect都是不错的挑选，惋惜当咱们开端构建Yelp数据管道时它们还没发布。因而就有了咱们的Schematizer。 Schematizer是什么? Yelp数据管道的一个重要规划便是将一切数据都形式化，也便是说，一切流经数据管道的数据都有必要恪守某种预先界说好的形式，而不是格局随意的。为什么非要着重这一点呢?由于咱们想让一切的数据顾客都能够对他们要获取的数据格局有预期，因而能够在上游数据出产者决议改动他们发布的数据形式时，不会对下流形成十分大的影响。一致的形式体现也让Yelp数据管道能够轻松地整合各种运用不同数据格局的体系。 Schematizer是用于盯梢和办理一切数据管道中用到的形式，并且供给主动化文档支撑等功用的形式存储服务。咱们运用Apache Avro来表达形式。Avro有许多咱们在数据管道中需求的功用，特别是形式演进，它是解耦数据出产者和顾客的关键因素之一。每一条流经数据管道的音讯都用Avro形式序列化过。为了减小音讯体积，咱们没有把悉数形式信息都放在音讯里，而仅仅带上了形式的ID。数据顾客能够用ID来在运行时从Schematizer中获取形式信息并将音讯反序列化。Schematizer是一切预界说的形式信息的仅有牢靠来历。 咱们用不同办法办理形式。 Schematizer用两种办法安排和办理形式：从数据出产者的视点和数据顾客的视点。 榜首种办法根据数据的产生信息来将形式分组，每个组由姓名空间和数据源来界说。出产者在向Schematizer注册形式时有必要供给姓名空间和数据源信息。比方一个预备向数据管道发布数据库数据的服务，它就能够把服务名作为姓名空间，把表名作为数据源。

根据姓名空间和数据源来将形式分组 第二种办法按数据的意图方信息来分组。比方Redshift集群或许MySQL数据库都是数据意图方，它们会对应一个或多个数据出产者，每个数据出产者又会相关一个或多个形式，这就对应着榜首种办法中界说的姓名空间和数据源。

根据单个数据意图方来将形式分组 这两种办法让咱们能够按不同的需求来检索和相关的形式。比方，一个程序或许想知道它会向哪些Topic发布数据，另一个服务又想知道它的Redshift集群中的数据都来自哪里。
咱们这样注册形式。 数据管道要求一切发布到其间的数据都有必要用预界说的Avro形式进行形式化和序列化。因而，当一个数据出产者预备向数据管道发布数据时，它要做的榜首件事便是向Schematizer注册形式，最通用的办法便是直接注册一个Avro形式。 关于没有或许无法创立Avro形式的数据出产者，也能够向Schematizer中参加形式转化器来把非Avro形式转化成Avro形式。MySQLStreamer便是一个代表，它是一个把MySQL数据库中的数据发布到数据管道的服务，它只知道MySQL表形式。Schematizer能够把MySQL表形式界说转化成相应的Avro形式。但假如数据出产者改动了形式界说的话，它有必要从头注册。 上游形式改动会不会影响下流服务? 一切数据管道服务都不能逃避的一同痛点便是该怎样应对上游形式改动。一般这都需求许多在上游出产者和下流顾客之间的交流和协调作业。Yelp也不能免俗。咱们也有批量使命和体系，它们要处理其他批量使命和体系产生的数据。每一次上游的形式改动都是十分苦楚的，它或许导致下流服务溃散，整个处理进程都是十分消耗人力的。 咱们经过形式兼容性来处理这个问题。在形式注册进程中，Schematizer会根据形式兼容性来决议Topic和新形式之间的对应联系。只需兼容的形式才干延用旧的Topic。假如有不兼容形式注册上来，Schematizer会用相同的姓名空间和数据源来为新形式注册一个新的Topic。那Schematizer又怎样确认兼容性呢?答案便是Avro解说规矩(Avro resolution rules)。Avro解说规矩确保在相同的Topic中，用新版形式打包的音讯能够按旧版形式解包，反之亦然。

不兼容的形式会分配不同的Topic 现在Yelp数据管道中大部分数据都产生自MySQLStreamer。比方咱们想为某事务表增加一个字段，MySQLStreamer就会向Schematizer注册新形式。由于依照Avro解说规矩这样的改动是兼容的，所以Schematizer会创立新Avro形式，并把这个姓名空间和数据源对应的旧的Topic分配给它。可假如是想把某字段从int改成varchar，那这便是一个不兼容的改动了，Schematizer会为新形式创立一个新Topic。 确保了在Topic内部的形式兼容性，下流数据顾客就能够定心的用旧形式去处理这个Topic中的任何数据，不用忧虑数据形式改动会引起本身的溃散等任何问题。他们也能够根据自己的需求在适宜的时分连上新Topic。这就让整个体系主动化程度更高，在形式改动时削减人工介入。 除了Avro解说规矩，咱们也在Schematizer中界说了一些自己的规矩来支撑一些数据管道功用。形式的主键字段被用于在数据管道中做日志紧缩。由于对同一个Topic来说做日志紧缩的主键有必要坚持一致，所以任何对主键的改动都被以为是不兼容的，会导致Schematizer为新形式创立一个新Topic。并且，当人工不可读(non-PII，Personally Identifiable Information)的形式开端包含人工可读字段时，这样的改动也被以为是不兼容的。人工不可读的数据和人工可读的数据必定分隔存储，这样就简化了人工可读数据的安全完结，避免了下流顾客不小心读到一些他们原本没有权限读的数据。

决议是否需求新Topic的逻辑流程 值得一提的是形式注册进程是幂等的。假如把相同的形式注册屡次，那只需榜首次会产生一个新形式，后边的都直接回来已注册的形式。这就让运用程序和服务能够十分简略地初始化它们的Avro形式。许多运用程序和服务都是把Avro形式界说在文件中或代码中的，但它们没办法写死形式ID，由于形式ID是由Schematizer管控的。所以运用程序能够调用形式注册接口来直接注册形式，假如现已存在就把形式信息取回来了，假如不存在就直接注册，一箭双雕。 将形式改动作业处理悉数流水线化。 为了让数据管道能够彻底以流水线的办法处理形式改动作业，Schematizer会根据当时形式和新形式的信息来为下流体系生成形式搬迁方案。现在Schematizer只能为Redshift表生成形式搬迁方案。关于把数据从数据管道中运用到Redshift集群的下流体系来说，在形式产生改动时它能够直接获取形式搬迁方案并且履行，并且主动获取新的形式信息，不需求任何人工介入。这个功用是很简略扩展的，并且形式搬迁方案生成器也是很简略替换的，所以将来咱们会增加更多的形式搬迁方案生成器来支撑更多的形式类型，或许改用更好的算法来生成搬迁方案。 Schematizer知道一切数据出产者和顾客的信息。 除了办理注册的形式，Schematizer还会跟进一切数据出产者和顾客的信息，包含哪个团队哪个服务担任出产或消费什么数据，发布数据的频率怎样，等等。在需求人工介入时咱们就能够用这些信息来有用地找到相应团队并与他们交流洽谈。并且这些信息也能够协助咱们监控和找出那些过期了的形式和Topic，然后能够将它们做废或删去。这样，就能够在新形式注册上来时简化兼容性验证作业。Schematizer能够越过那些抛弃的形式，只检查新形式与Topic内剩余的有用的形式的兼容性就能够了。 一切数据出产者和顾客在启动时都有必要供给这些信息。开始咱们只想着把它们保存在Schematizer里就好了，但事实上这些信息对探索性的剖析和预警都是十分有用的，终究咱们决议把它们写到数据管道体系之外的独自的Kafka Topic中。这样数据就能够被Redshift和Splunk处理，也能够导入Schematizer和经过前端Web界面展示出来。咱们用的是Yelp自行研制的经过Clog写入数据的异步、非堵塞式Kafka出产者，这样就不会影响出产者正常地发布数据。别的，这样也能够避免环形依靠，有时分正常的出产者要用相同的信息去注册屡次。 该用哪个Kafka Topic呢?Schematizer会处理好这些细节。 与一般含义上的Kafka出产者不同，数据管道的数据出产者不需求事前知道它们应该把数据发送到哪个Kafka Topic中。由于Schematizer规则了注册上来的形式和Topic之间的对应联系，所以数据出产者只需供给自己序列化数据所运用的形式信息，就能够从Schematizer那里得到正确的Topic信息并发布数据了。将Topic信息笼统出去能够让接口更简略易用。 对数据顾客也是相似的机制。虽然也能够给它们定下一些详细的Topic去消费，但更常见的用例是让Schematizer根据数据顾客感兴趣的组的信息来供给正确的Topic。在本文前面章节介绍了各种不同的分组机制。数据消费都能够或许指定姓名空间和数据源，或许指定数据意图方，Schematizer就会找出那个组内的相应Topic。这种机制关于数据顾客感兴趣的一组Topic或许由于形式的不兼容改动而变来变去的场景特别有用。它让数据顾客不用再盯梢组内的每一个Topic。 形式很好，文档更好! 形式把数据格局化了，但关于想了解数据切当含义的人来说供给的信息或许又不行。咱们注意到运用数据的人一般不是出产数据的人，因而他们不知道去哪里找到有用的信息来让他们了解他们要用的数据。由于Schematizer担任办理数据管道中的一切形式，所以把数据的描绘信息也保存在它这儿就很适宜。 常识发掘器Watson盛大进场。 Schematizer要求形式的注册方跟着形式一同供给文档，然后Schematizer会提取文档信息并保存起来。为了让Yelp公司内的各个团队能够获得形式和数据文档，咱们开发了Watson，一个全公司职工都能够用来发掘数据内容的Webapp。Watson实际上是Schematizer的一个可视化前端，它经过Schematizer的几个RESTful API来获取数据。 Watson供给了关于数据管道状况的有价值信息：现有的姓名空间、数据源及相关的Avro形式信息。最重要的是，Watson为检查Schematizer办理的一切数据源和形式信息供给了简略的办法。 文档并不是天上掉下来的。 现在流经咱们数据管道的数据首要都来自于数据库。咱们用SQLAlchemy模型来为这些数据的数据源和形式收拾文档。在Yelp，SQLAlchemy用来描绘咱们数据库中的一切模型。除了docstring之外，SQLAlchemy还答运用户为模型的字段增加额定信息。因而，它天然成了咱们保存文档的首选之处，记载各个数据模型和字段的意图和含义。 SQLAlchemy还引入了一个属主字段来记载每个模型的维护者和专家。咱们以为生成数据的人是供给文档的最佳人选。别的，这种办法也会鼓舞咱们时间坚持实在数据模型与描绘的同步。

classBizModel(Base):
__yelp_owner__=Ownership(
teams=[TEAM_OWNERS['biz_team'],
members=[],
contacts=[]
)
__table_name__='my_biz_table'
__doc__='Businessinformation.'
id=Column(Integer,primary_key=True,doc=r"""IDofthebusiness.""")
name=Column(String(64),doc=r"""Nameofthebusiness.""")

一个简略的包含文档和属主信息的SQLAlchemy模型 但是开发者在做SQLAlchemy模型的时分并不总是会记住供给文档信息。为了避免这样的作业产生，咱们开发了主动校验功用来强制要求一切模型都有必要完整地供给了特点描绘和文档，这是绝不会让步的硬性规范。每逢有新模型要参加时，假如要求的文档信息不齐备，或许没有属主信息，校验就会失利。这些主动校验功用协助咱们朝着100%文档覆盖率的方针迈进了一大步。 为Watson提取高质量文档。 当数据模型有了文档之后，咱们就能够把它导入Schematizer并终究经过Watson展示出去。在深化详细提取流程之前，咱们先介绍一下这个进程中的另一个重要模块：特定运用转化器(Application Specific Transformer)，简写为AST。与姓名含义相同，AST从一个或多个数据管道Topic中输入音讯流，用转化逻辑处理音讯形式和数据包，再把转化后的音讯输出到别的的数据管道Topic中。供给详细转化处理的转化模块是能够串连起来的，因而能够组合多个模块来做十分详尽的转化作业。 咱们用AST中的许多个转化模块来根据SQLAlchemy模型生成更易了解的数据。由于模块是能够串连的，现在咱们仅仅简略的创立一个从SQLAlchemy模型中提取文档和属主信息的转化模块，并把它参加 到已有的转化链中。这样，一切模型的文档和属主信息就经过现有管道主动提取并导入Schematizer了。完结进程适当简略，并无缝接入管道，所以能够十分有用地生成高质量文档。

AST中的转化模块 如上所述，AST中现在现已有了一些为用户生成更有含义的信息的转化模块。位标志转化模块会解说一个整型字段的不同数据位的详细含义。相似地，Enum字段转化模块也会把Enum值转化成可读的文字表述。这些转化模块带来的另一个优点是它们一同也产生了自解说和自生成文档的形式，因而也产生了更好的文档。 协作、奉献与检索 开发者的文档并不是咱们要叙述的最终一项内容。Watson也供给了功用让终端用户能够一同尽力，为使Yelp的数据更具可读性而奉献自己的力气。 榜首个功用便是打标签。Watson答运用户为恣意数据源打标签分类。一个数据源或许是个MySQL数据库表，也或许是个数据模型。比方，一个事务数据源能够打上“Business Information”的标签，而一个用户信息数据源能够打上“User Information”标签。终端用户能够把相关的数据源都打上相同的标签，这样以对自己最有含义的办法把它们安排在一同。打标签能够让咱们更深化的了解咱们的各个数据源之间是怎样互相相关的。

打上了“Business Info”标签的事务数据源 Watson供给的另一个功用是增加注释。终端用户，特别是非技术人员，能够经过这种办法来为一个数据源或字段供给他们自己的文档。比方事务剖析师常常就会对运用数据有十分名贵的见地，他们能够经过注释来共享各种疑难杂症、鸿沟用例和时效性很强的信息。 终端用户关于Watson的最大的需求便是检索。咱们在Watson中完结了简略的检索引擎，让用户能够检索数据的形式、Topic、数据模型描绘等各方面信息。在检索后台咱们没有用Elasticsearch，而是挑选了Whoosh Python包，由于它能够协助咱们快速完结开发。就咱们现在的检索量来说Whoosh的功用足以敷衍。跟着数据规划增大，咱们将来会考虑换用其它更易扩展的引擎。 定论 Schematizer是Yelp数据管道的一个重要组成部分。它的形式注册操作是数据管道的许多重要功用的根底，包含在上游数据出产者更改形式时减轻对下流顾客程序和服务的影响等。Schematizer也办理了数据发布的Topic分配，让用户不用再关怀详细运用哪个Topic等这样的细节。最终，它要求一切写入数据管道的数据都有必要有文档，这促进了全公司内的常识共享。Watson的参加更是使得Yelp公司内的一切职工都能够方便地得到最及时的信息。