1、Apache InLong的 SPI 扩展实践周康 腾讯数据 级开发程师Apache InLong PMC 成员Email:?healchowapache.org|0101Apache?InLong?Apache?InLong?简介简介项简介适场景0303InLongInLong ManagerManager 的的 SPISPI 改造实践改造实践存在的问题什么是 SPISPI?改造实践录 CONTENT|0202InLong?ManagerInLong?Manager 简介简介Manager 的作0101Apache?InLong?Apache?InLong?简介简介项简介适场景|01 Apa

2、che?InLong?简介/是什么官网：https:/inlong.apache.orgApache InLong（应龙）是一个一站式的海量数据集成框架，提供自动、安全、可靠和高性能的数据传输能力，方便业务构建基于流式的数据分析、建模和应用。表单数据日志Apps 数据数据集成PrestoHiveMapReduceHDFSHBaseS3SparkFlinkAd-hoc 查询BI人物画像数据源数据仓库数据应用ES最初于 2019 年 11 月由腾讯大数据团队捐献到 Apache 孵化器，2022 年 6 月正式孵化毕业，成为 Apache 顶级项目（TLP）。|01 Apache?InLong?简

3、介/适场景|依托腾讯百万亿级别的数据接入和处理能力，整合了数据采集、汇聚、缓存、分拣全流程，具有简单易用、稳定可靠、灵活扩展等特性。广泛应用于广告、支付、社交、游戏、运营商、人工智能等领域。|0202InLong?ManagerInLong?Manager 简介简介Manager 的作|02 InLong?Manager 简介|InLong 支持数据的采集、汇聚、缓存和分拣，只需一些基础配置，就可把数据从源端导入到实时计算引擎或者写入离线存储系统。系统如何将这些流程串联起来？|通过 InLong Manager 来管理系统和任务的元数据，串联任务的全流程。元数据主要有：审批信息 集群配置信息

4、数据 schema 配置（源、目标）02 InLong?Manager 简介|通过 InLong Dashboard 提供的 Web UI（或 Manager Client 提供的命令行工具）创建数据流任务，任务审批通过后，即可串联起全部流程，主要包括：|1.创建目标端的库表结构2.创建 MQ 的 Topic 和消费者3.启动 Flink 任务，开始从 MQ 消费数据，写入目标端4.下发采集任务，向 MQ 生产数据0303ManagerManager 的的 SPISPI 改造实践改造实践什么是 SPI改造过程改造后的收益|03 Manager 的 SPI 改造实践/存在的问题|InLong 源

5、于腾讯内网业务，在近10年的发展中，主要支持的数据源和数据存储如下：|数据源端数据目标端03 Manager 的 SPI 改造实践/存在的问题|以数据存储端为例，由于用到的存储类型有限，且考虑到不同的存储类型的参数差异较大，我们的配置表是这样设计的：|03 Manager 的 SPI 改造实践/存在的问题|在 InLong 上云的过程中，数据源端和目标端的类型急剧增多。可以预见，随着云上客户规模的增加，还会继续扩展更多类型的数据源和目标端。|数据源端数据目标端在前面已有类型的基础上，新增的类型03 Manager 的 SPI 改造实践/存在的问题|扩展的过程中发现的痛点：维护成本高：表多，重复

6、字段多 大量相似代码（if-else/switch-case 处理相似逻辑）难扩展：要扩展新的存储类型，不仅要添加一张表，还要修改接口中的代码，添加 else/case 语句（不符合开闭原则）|03 Manager 的 SPI 改造实践/什么是 SPI|SPI，全称 Service Provider Interface，是 Java 提供的一套用来被第三方实现或者扩展的 API，它可以用来启用框架扩展和替换组件。翻译后是“服务提供者接口”，顾名思义，这个接口是给“服务提供者”使用的。|常见示例：加载 数据库驱动 load 接口的实现类SLF4J 加载不同提供商的日志实现类 日志门面接口的实现类

7、Spring 中自动类型转换 Type Conversion SPI(Converter SPI、Formatter SPI)等03 Manager 的 SPI 改造实践/什么是 SPI|以 Flink JDBC Connector 中对不同 JDBC 方言的处理为例：1）首先定义一个开放给外部去实现的 JdbcDialectFactory 接口，由不同的 DB Dialect 去实现：|03 Manager 的 SPI 改造实践/什么是 SPI|2）在classpath的META-INF/services 目录下创建一个名为此接口全限定名的文件，内容是各个实现类的全限定名：|3）程序中通过j

8、ava.util.ServiceLoder扫描META-INF/services目录下的配置文件，根据实现类的全限定名来动态装载具体的实现类：03 Manager 的 SPI 改造实践/什么是 SPI|只需要调用 3）中JdbcDialectLoader的加载方法，即可根据传入的参数动态选中符合条件的实现类，比如传入的 JDBC URL 以 jdbc:mysql:开头，在JdbcDialectLoader中就会返回 MySqlDialectFactory：|Java SPI 实际上是“基于接口的编程策略模式配置文件”组合实现的动态加载机制。public static JdbcDialect l

9、oad(String url)ClassLoader cl=Thread.currentThread().getContextClassLoader();List foundFactories=discoverFactories(cl);/.final List matchingFactories=foundFactories.stream().filter(f-f.acceptsURL(url).collect(Collectors.toList();/.return matchingFactories.get(0).create();03 Manager 的 SPI 改造实践/改造过程|1

10、）精简服务层代码，删除繁琐的 if-else/switch-case。收敛 Service 层的接口，同一领域模型的请求都在同一个 Service 接口中处理，以保存操作为例：|代码路径:org.apache.inlong.manager.service.sink.StreamSinkServiceImpl03 Manager 的 SPI 改造实践/改造过程|根据不同的类型，通过执行器的工厂找到具体的配置执行器，该执行器去执行真正的保存方法：|代码路径:org.apache.inlong.manager.service.sink.SinkOperatorFactory03 Manager 的

11、SPI 改造实践/改造过程|2）重构数据库实体模型，一张表支持任意类型 Sink 的配置。表中记录通用字段，对于不同类型 Sink 的特有字段，通过一个扩展字段来存储（KV，JSON）：|03 Manager 的 SPI 改造实践/改造过程|不同类型 Sink 的特有参数，转换成 JSON 格式存储到ext_params字段中，查询时再将其解析成其特有的 DTO。相关代码可以参考：org.apache.inlong.manager.service.sink.AbstractSinkOperator 中 saveOpt 方法调用的 setTargetEntity 方法org.apache.inl

12、ong.manager.service.sink.StreamSinkOperator 接口中getFromEntity 方法的各个实现|其他类似操作，感兴趣的朋友请检视：org.apache.inlong.manager.service.group.InlongGroupOperatororg.apache.inlong.manager.service.sink.StreamSinkOperatororg.apache.inlong.manager.service.source.StreamSourceOperatororg.apache.inlong.manager.service.res

13、ource.sink.SinkResourceOperatororg.apache.inlong.manager.service.resource.queue.QueueResourceOperator03 Manager 的 SPI 改造实践/改造后的收益|在经过上述改造后，至少带来了如下收益：1）代码复用性提高，减少了大量重复/相似逻辑的代码，降低了维护成本2）代码扩展性极大增强，增加不同类型的配置，只需要依葫芦画瓢，去实现其特殊逻辑即可，也无需改动已有接口3）表的 DDL 不用频繁变动，降低了维护成本，避免修改 DDL/增加新表 引发线上问题4）可以在不侵入修改开源代码的情况下，扩展腾讯内部的配置类型，加入内网特有的业务逻辑。|