Salesforce连接器在Yelp中的应用案例

Yelp是美国著名商户点评网站，创立于2004年，囊括各地餐馆、购物中心、酒店、旅游等领域的商户，用户可以在Yelp网站中给商户打分，提交评论，交流购物体验等。

在Yelp 中搜索一个餐厅或者旅馆，能看到它的简要介绍以及网友的点论，点评者还会给出多少星级的评价，通常点评者都是亲身体验过该商户服务的消费者，评论大多形象细致。

更好理解的解释是Yelp就是中国版的大众点评，巧合的是大众点评也在用Salesforce来管理商户。

Yelp用客户关系管理（customer relationship management，CRM）平台Salesforce来为超过2000人的销售团队提供支持。Salesforce提供了许多现成的功能，让我们的销售团队可以轻松地定制他们的业务流程。

Yelp的销售团队主要做什么呢？他们卖广告包！他们卖给谁呢？Yelp上的商家！那我们如何从Yelp的数据库中获取这些商家信息并提供给Salesforce呢？读过下文你就明白了。

以前的方法

我们现有的单向同步基础架构名为“Bulk Workers”，是早在2010年设计的了，目的是要显著地改进端到端发送数据的时间。这套设计方案成功地把同步时间从3星期缩短为24小时，这很棒！这套方案细节是怎样的呢？

这些“Bulk Workers”实际上就是一些Gearman定时任务，它们会检索业务表中的每一行数据，再按Salesforce的模式对这些数据进行转换。转换后的数据会用一个Salesforce客户端发送给Salesforce，客户端是基于Beatbox开发的，主要的改进是增加了对Salesforce批量API的支持。

旧的集成方法

上面的方法一直工作得很好……直到它再也顶不住的时候。

在接下来的5年里，这套架构见证了数据的急速增长，从最初的30万亿行，到2015年超过了100万亿行。数据更新的操作影响尤其大。后来，同步时间开始越拖越长。我们就知道我们需要优化一下系统，来更好地支持更新操作了。

进入正题：数据管道

于是我们开始收集需求。我们认为新的解决方案需要下面这些：

• 实时处理
• 保证“至少一次提交”
• 自带监控和告警等功能
• 由配置驱动模式之间的转换
• 可以很容易地增加新字段和转换

差不多是在相同的时间点，我们已经在做依靠分布式发布/订阅消息系统Kafka的新数据管道了。这个数据管道可以直接满足我们的前三个需求。那我们剩下的工作就是构建一个转换框架来满足后面的两个需求，并且作为到Salesforce.com的连接器。

Salesforce数据管道集成方法

转换器（Transformer）

我们采用了一个名为PaaStorm的、在Yelp Hackathon上产生的并且已经应用到生产环境的项目作为我们的Kafka-to-Kafka处理器，叫这个名字是因为它与Storm很相似，并且是用Yelp PaaSTA部署的。在保留了Storm的范式的前提下，我们构建了一个通用的转换器，可以生成许多实例，处理各个Topic中的要发往Salesforce的原始数据。在处理源Topic时，每个实例都会从一个YAML文件中得到转换步骤，然后再做拷贝、移动和（或）值映射操作。这很重要，因为Salesforce的模式存在的时间远远长于我们的新架构，是很难改动的。这个也意味着没有什么比较自动的方法去做字段映射。如果有一个由配置驱动的映射关系，那就让我们可以快速对转换做解析而不必真正地部署代码。这对项目的灵活性至关重要。

每个转换器都会向一个新的Kafka Topic中发布序列化后的Salesforce对象，供上传器消费并发往Salesforce。

上传器（Uploader）

让上传器自己做为一个实例就可以让我们清楚地知道我们有多少个服务要与Salesforce.com通信。上传器会消费各个转换器转换后的消息，将它们批量发送到Salesforce。因为发往Salesforce的请求是发向互联网的，所以这是我们的管道中最慢的部分之一。因而能不有效地进行批量处理就对性能影响非常大。使用合适的API也非常重要。因为Salesforce提供了好多种不同的API，所以有时候选择用哪个也不容易。为了让我们可以很容易地不必做更多的工作就在API之间切换，我们写了个通用的客户端，用它包装了现有的SOAP、REST和批量API等Python客户端。我们也写了一个ORM（Object-Relational Mapping）客户端，并为每张我们要写的目标表定义了模型。这让我们可以在发送到Salesforce.com之前就对数据进行验证，并且在写数据的时候确认该用哪个Salesforce External ID。

评估

我们用来评估的第一张表就是我们的广告商表。选它是因为它是我们Yelp整体业务中规模比较小的表之一，但对于我们的销售团队的运作又是非常重要的。在之前要把数据变更操作从Yelp同步到Salesforce需要16小时。在切换到新架构之后，可以看到同步时间变成了平均约10秒钟，偶尔会有一些波峰，但也就只是变为几分钟而已！这就足以表明出现在Salesforce上的数据是非常可靠的，销售团队在查看数据时，再也不必不断地来要更新的数据了。

每秒钟平均同步延迟

挑战

听起来很容易？事实上并非如此。在设计和构建我们的连接器的过程中，我们要解决好几个问题：

我们注意到在处理失败的更新操作中有很大部分都是在Salesforce一侧超时了，或者是由于没能成功的为某行数据获取锁而被拒绝了。这两种问题的根本原因都在于我们在Salesforce的程序中使用了大量的触发器和回滚操作。差不多每张表上都有非常复杂的逻辑，而每一条写操作都要把这些逻辑全处理一遍，以保证不同数据之间的一致性，或者为了将某些业务流程自动化。这些功能本来都是很好的，但碰上问题时你就不那么想了。所以我们工作的重点就是要减少做写操作时的处理量。把这样的处理尽可能地挪到异步处理的过程中，就可以减少我们锁定单条记录的时间，也就减少了每条写操作的处理时间。

另一个要解决的问题是依赖关系。我们本来的数据源（MySQL）有限制依赖，而Kafka并没有。虽然写到每个Kafka Topic中的消息都是保证有序的，但是我们并不能保证这些Topic中的数据会以某个确定的速度被处理。在各张表都彼此依赖的情况下这个问题就很严重，因为一张表中的数据可能会比另一张表的数据更先被读取和更新，导致数据在一定时间内处于不一致的状态。一个常见的例子就是广告商的数据记录会比用户的数据稍早到一会。因为广告商的数据中包含一个指向User表的外键字段，写入就会失败。因此我们就要跟踪哪些数据是因为不符合依赖约束而写入失败的，然后再由上传器在确定依赖关系满足了之后再重试。把上传的操作按依赖顺序序列化并处理重试，这可以覆盖我们绝大多数的用例，尽管这意味着我们因此没办法达到一个很高的并行度。

还有一个问题，就是我们的数据没有全放在一个单一的数据库里面，所以对我们来说可用的就是单条的数据记录。为了解决这样的问题我们开发了新功能，读两个Topic中的数据并把它们关联起来，然后再把关联后的数据重新发布出去。

结论

使用基于Kafka的数据管道来为销售团队获取数据，我们已经在这方面取得了很大改进。接下来我们准备构建自己的基础架构，这样就可以实现其他的转换操作、简单的聚合、以及在写Salesforce的高可靠保障等等功能。