2022年第011篇
本文上篇介绍了一下neutrino的原理和基本用法，下篇介绍扩展功能，包括如何利用neutrino传递任意对象，以及checkpoint恢复，对象生命周期控制，和其他扩展功能。
如果一定要绑定并传递不可继承类如final class，neutrino提供了一个底层API：SerializableProvider接口，可实现无缝传递。
使用过Spring或Guice API的读者一定对Provider接口比较熟悉，SerializableProvider实际上就是一个同时包含Serializable和Provider的接口。
例如，有一段处理点击事件的代码，如下所示：
如果使用前面同样绑定方式，则编译报错，因为无法为final类继承生成一个Proxy。
那么这种情况可以改用如下的方法：
① 绑定final类EventProcessor
② 绑定EventProcessor的SerializableProvider，提供SerializableProvider[EventProvider]注入
那么如何生成EventProcessor实例呢？
这里的Provider包含对应依赖图节点的id，可实现无缝传递。事实上，上篇中自动生成的代理也是基于SerializableProvider实现的。
当然SerializableProvider也可以通过构造函数注入的方式使用，比如：
我们可以看到，neutrino的对于对象传递时序列化的处理就是在worker上利用依赖关系图重新创建对象。这带来一个问题，即如果依赖注入生成对象的状态在driver上被修改，那么这种修改无法通过这种方式传递到worker上。
如果开发者一定要用neutrino框架生成可变状态的对象，可以使用guice原生的方式绑定（即不使用withSerializableProxy或者SerializableProvider绑定），但是仍然可以依赖于以neutrino的API绑定的对象。
例如我们把上篇中的例子修改一下：为了减轻对后台的压力，提供item信息的API服务为后台加了缓存。并在API加了一个参数（1~5之间），用来指示缓存级别。1表示绕过缓存，直接读取后台数据，对后台压力大；5表示1小时缓存，对后台压力小，但数据可能是旧的。平时我们可以采用级别1（即不读缓存）；大促销期间压力大，可以通过配置动态改变缓存级别参数，而不需要重启服务。
在具体实现中，我们在spark streaming服务的每一个时间间隔从配置中心读取当前的缓存级别。代码如下：
可以看到此时HttpEventEnrichment增加了一个成员变量用来表示当前缓存级别，该变量在driver上初始化时从配置中心读入。这使得enrichment对象的状态在从依赖注入容器生成之后发生了改变，而该变化又必须被传递到worker上，因时EventEnrichment不能再使用neutrino的绑定方式（即使用withSerializableProxy和SerializableProvider），此时它将作为普通java对象传递，实现了Serializable接口。以下是EnrichmentModule的代码：
此处EventEnrichment按照Guice原生API绑定，这样enrichment不再是代理，而是HttpEventEnrichment实例。
而且由于HttpEventEnrichment依赖于HttpClient，后者实际上会随着enrichment对象被传递到worker，所以我们需要调用withSerializableProxy为HttpClient生成一个代理，保证在worker上利用当前JVM的依赖图生成一个实例。也就是说Java原生序列化对象也可以依赖于neutrino产生的代理或者SerializableProvider。
注意此时因为enrichment对象每个时间间隔的状态可能会发生变化，scope不能再设置为Singleton，此时修改为StreamingBatchScope，即每一个spark streaming周期内共用一个实例。或者可以使用Guice默认的NO_SCOPE，即每次请求自动生成一个新实例。
不过我本人并不推荐通过依赖注入生成可变状态对象。上面的例子中，读取缓存级别实际上可以看成创建EventEnrichment的中间步骤，所以也可以通过抽象工厂模式实现。代码如下：
这种就可以规避对象状态可变的问题。
一般情况下，只在driver上使用的对象不需要被序列化。但是如果使用spark的checkpoint功能，则其中有些对象也需要被序列化。neutrino框架对于这部分也有相应支持。
例如使用上文中提到的RDDHandler，如果改写成支持checkpoint形式，代码如下：
①在创建StreamingContext之前创建injector，上文已介绍
②在spark streaming任务中使用checkpoint功能时，需调用neutrino提供的sparkSession.getOrCreateStreamingContext方法，传入的回调函数会在没有checkpoint数据存在时被调用
③创建StreamingContext，此处不能直接调用StreamingContext构造函数，而应该调用neutrino提供的session.newStreamingContext方法
④设置checkpoint地址
⑤从injector获得handler实例
一般情况下，如果使用了checkpoint，虽然handler对象只在driver上使用(foreachRDD中的回调在driver上执行)，但是因为参与了RDD DAG的构建，从checkpoint恢复重建DAG时需要用到这个handler对象，所以它也要被序列化。
但是，如果使用了neutrino这个问题可以不用考虑，唯一需要做的是在绑定RDDHanlder的时候调用withSerializableProxy
neutrino会在spark任务从checkpoint恢复时自动根据依赖关系图重建这个对象。
实际上，neutrino在spark任务从checkpoint恢复时，首先恢复各个JVM上的依赖图，然后根据该图重新创建driver和worker上的对象。
neutrino提供了关键spark对象的注入功能，如SparkSession,SparkContext, StreamingContext都支持注入，当然前提是它们只能在driver上使用。
例如我们可以用这一功能抽象一个Stream的生成：
绑定之后就可以直接从injector得到DStream[ClickEvent]实例：
熟悉Guice和Spring API的读者可能知道，guice injector （或Spring的ApplicationContext）支持继承，即父子injector (parent/child injector)，本框架对此也提供支持。具体API用法见开源项目README，在此不再赘述。
下篇首先介绍使用底层API实现任意类型实例（包括final类型绑定实例）的传递，接着综合上篇提到的上层API（自动代理）方法，分析了neutrino的使用限制，以及解决的方法。最后提到了neutrino的一些其他的功能。
Russell Bie，内容发现部门资深软件工程师
内容发现部门(Content Discovery Org.)是迪士尼流媒体核心研发部门，主攻Hulu、Disney+、Star+等迪士尼流媒体产品线的三大业务方向：搜索、个性化推荐、内容推广。在每个业务方向上都和人工智能技术深度融合，涉及AI平台的搭建、前沿算法的研究、以及工程系统的集成，致力于为迪士尼流媒体用户提供最佳的视频观看体验。
自成立伊始，该部门始终将内容的精准传递作为首要业务目标，深入结合工程、算法和数据，利用人才优势与人工智能基础解决业务问题。
职位信息: 职位列表链接
感兴趣的同学发送简历至：bjindustry@disney.com
（烦请标注申请职位+姓名）
