做不到实时

做不到实时的原因

实时计算很难。通过增量计算的方式来间接获得问题的（伪）实时结果，即使这些结果带有迟滞性和近似性，但只要能够带来尽可能最新的信息，那也是有价值的。

原因可分成3个方面：

1. 算法复杂度高：待解决问题很复杂。如风控分析，需统计社交网络中二度联系人之间的关系。这类问题的复杂度通常会大于$O(N)$。计算时延大导致系统非实时；
2. 计算资源受限：当计算资源中的一种或多种已经用完时，则计算任务会出现排队等待的情况，这会增加处理的时延。监控的必要性：事故预警、性能优化指出方向。
3. 数据量过大：当数据量过大时，即使是时间复杂度为$O(logN)$级别的算法，也可能因数据分布在多个节点上，需跨主机远程访问，带来过多I/O操作；数据量过大，给存储管理带来复杂性，还对计算造成影响，如频繁GC造成JVM卡顿。

木桶原理：一旦流计算系统的某个环节非实时，则整个系统的处理速度就会受限于这个非实时环节。

前面也提到过，针对系统瓶颈（木桶短板）进行优化。

Lambda架构

在设计数据系统时，千万不要将查询和更新两个过程耦合起来。

当数据量太大而不能实时全量计算时，Lambda架构将数据处理过程分成两部分：

• 基于批处理的预计算：
• 基于流处理的实时增量计算：

Lambda架构将数据系统视为在不可变数据集上的纯函数计算，这与函数式编程的核心思想是不谋而合的。

Lambda架构总体上分为3层：

• 批处理层：Batch Layer，存储主数据集和预计算各种批处理视图；
• 流处理层：Speed Layer，实时计算并存储在批处理层两次调度执行期间新增数据；
• 服务层：Serving Layer，将批处理层和快速处理层的计算结果合并起来，为用户实时提供全量数据集上的查询。

Lambda架构是一种思想，每一层的组件选型没有严格限定。

在实时流计算中运用Lambda架构

离线计算用于训练更新模型参数，实时计算用于进行在线风险评分。

Kappa架构

Lambda架构最主要的问题，对于同一个查询，需分别为批处理层和快速计算层开发两种不同的代码，给开发测试和运维都带来复杂性和工作量。

Kappa架构的核心思路是将批处理层用快速处理层的流计算技术替换。

流计算和批处理，流数据和块数据，其实没有什么界限：流是由块构成。

Kappa架构如下

Kappa架构本质上依旧是Lambda架构，只是原本用作离线的批处理层被流计算取代。

采用Kappa架构，只需要一套代码，不同的仅仅是两个时间窗口的类型：

DataStream counts = stream
	.map(new MapFunction<String, Event>() {
		@Override
		public Event map(String s) throws Exception {
			return JSONObject.parseObject(s, Event.class);
		}
	})
	// 事件转化
	.map(new MapFunction<Event, CountedEvent>() {
		@Override
		public CountedEvent map(Event event) throws Exception {
			return new CountedEvent(event.product, 1, event.timestamp);
		}
	})
	.assignTimestampsAndWatermarks(new EventTimestampPeriodicWatermarks())
	.keyBy("product")
	// 批处理层，使用滑动窗口SlidingEventTimeWindows
	.timeWindow(Time.days(3), Time.minutes(30))
	// 流处理层，使用翻转窗口TumblingEventTimeWindows
	// .window(TumblingEventTimeWindows.of(Time.seconds(15)))
	// 聚合计算
	.reduce((e1, e2) -> {
		CountedEvent event = new CountedEvent();
		event.product = e1.product;
		event.timestamp = e1.timestamp;
		event.count = e1.count + e2.count;
		event.minTimestamp = Math.min(e1.minTimestamp, e2.minTimestamp);
		event.maxTimestamp = Math.min(e1.maxTimestamp, e2.maxTimestamp);
		return event;
	});

小结

Lambda架构是一种构建数据系统的思想，将数据分析的过程定义为在不可变数据集合上的纯函数计算。数据系统的构建过程分成两步：收集一批数据形成不可变数据集、在不可变数据集上进行数据处理和分析。这种思路，可应用于离线、实时处理部分。分别对应Lambda架构的批处理层和快速处理层。

由Lambda架构演进而来的Kappa架构，通过流来统一编程界面，极大地简化数据系统的构建过程。在架构体系和代码开发过程中，Kappa相比Lambda具有更好的一致性。但这并不意味着Kappa比Lambda架构更好，它们有各自的意义和价值。Lambda架构代表的是一种更通用的架构思想，在碰到不能直接用实时计算方式解决大数据问题时，不妨尝试采用这种离线和实时相结合的折中方案。Kappa架构的最大价值则是尽量用流式计算框架来统一离线计算和实时计算。

数据传输

消息中间件

功能

MQ的功能：

• 将上下游业务逻辑处理单元解耦
• 缓冲消息和平滑流量高峰
• 使系统的处理能力能够横向扩展
• 消息高可靠传递
• 消息的分区和保序

分区和保序：用分区消息局部有序性来取代全体消息整体有序性，在很多业务场景下都能够满足对消息顺序的要求，同时不会影响处理性能的水平扩展。

工作模式

包括：

• 点对点模式：Point-to-Point，P2P，一条消息只能由一个消费者消费；
• 发布/订阅模式：订阅组的负载均衡；

消息模式

包括：

• 无模式：弱模式，如JSON，无schema编程；数据不变性原则：加字段，不能改或删字段；
• 强模式：定义严格的数据模式
• 版本控制

注意事项

两个：

• 可靠性：MQ承诺的精确一次会非常显著地降低MQ组件性能；不同MQ承诺的精确一次语义或多或少有所区别，使用起来还有一定限制，如Storm的ACK机制和Trident、Flink的checkpoint机制和Keyed State及Operator State等。
• 消息重放：因业务Bug或其他问题，需重新消费消息。Kafka提供Offset调整机制。

Kafka

Kafka是一种全新的数据管理方式，可直接以流的方式来存储、查询和管理数据。

对于存储在broker上的消息，都会设置一个超期策略，可按时间超期或按数量超期，超期的消息会被淘汰。任何类型的线上数据都不会永远有效，两点含义：在设计数据存储系统时，务必设置一个超期淘汰机制；在为业务设计实体关系模型时，应该认真考虑数据在业务意义下的真实有效时间。

配置KafkaProducer时需考虑四个方面：

• 消息的可靠性：ack三个参数设置；
• 同步或异步：producer.type=async；
• 批次发送：batch.size可控制批次发送的消息数量，而lingger.ms则可控制收集消息的时间；
• 压缩：压缩有一定CPU开销，但有两个好处：减少消息发送时的网络流量和磁盘空间。

消费者和消费者组：同一主题的消息能够被多个消费者组消费，各个消费者组相互独立，互不影响。但在同一个消费组里的消费者，则齐心协力共同处理同一主题下的消息，当一个消息被一个消费者认领后，同一个消费者组里的其他消费者就不再认领该消息，这样就保证了能够横向扩展并行处理的消费者数量。

消费者和分区：在同一个消费者组内，任何一个分区在同一时刻都只允许有一个消费者负责读取该分区中的消息。

配置KafkaConsumer时需考虑三个方面：

• 消费者组内消费者的数量：由于一个分区只能被同一消费者组内的一个消费者读取，而一个消费者可以读多个分区的数据，所以配置超过分区数的消费者数量并不会提升主题中消息处理的速度；
• fetch.min.bytes和fetch.max.wait.ms：分别决定消费者一次读取消息的条数及最多等待Kafka broker将数据收集全的时间。当消息不足fetch.min.bytes定义的字节数，而时间达到fetch.max.wait.ms时，broker会将已经收集到的消息一次性返回给消费者。很明显，这种设计也是为了减少I/O次数，提高每次消息有效载荷，从而提高消费者读取消息的性能；
• checkpoint时间间隔：消费者侧有保证消息可靠性读取的机制。这就是replica.high.watermark.checkpoint.interval.ms参数的功能。当从分区读取出消息后，可以将本次读取消息的偏移量提交到ZooKeeper保存下来。当后续因为处理失败等原因，需要重新处理消息时，直接跳回标记点重新读取消息即可。如果每次都设置一个checkpoint，那么我们将永远不会丢失消息，但是这样做会明显地影响消费者性能。如果我们隔一段时间或对一定数量的消息数设置checkpoint，就可以在性能和可靠性之间获得一个合适的平衡点。

Kafka三大特点：超高吞吐率、高可水平扩展性及能够直接高可靠地存储流式数据，可当做数据总线使用。

RabbitMQ

RabbitMQ严格遵守AMQP标准，提供如下组件：

• Broker：RabbitMQ的服务节点，多个Broker能够组建为集群；
• Vhost：虚拟主机，是对Broker的逻辑划分，可以实现诸如资源隔离和用户权限隔离的功能；
• Exchange：消息交换器，用于将消息按照设定的规则路由到一个或多个队列；
• Queue：消息队列，用于暂存由Exchange投递的消息；
• Binding：绑定，相当于Exchange的路由表，将Exchange和Queue按照设定的路由规则绑定起来；
• RoutingKey：路由主键，Exchange在执行路由规则时使用的主键，是一个消息头；
• BindingKey：指定哪些RoutingKey会被路由到相应Exchange绑定的Queue中；
• Producer：消息生产者，指发送消息到Broker的客户端程序；
• Consumer：消息消费者，指从Broker读取消息的客户端程序；
• Connection：与RabbitMQ服务器的连接；
• Channel：消息通道，构建于Connection上的通道，是与Exchange或Queue的连接，一个Connection上可以构建多个Channel；

将RabbitMQ用于配置总线：Spring Cloud Config + Spring Cloud Bus + RabbitMQ。

Camel

为了能够更好地管理和维护底层的消息中间件，让其能够平滑地跟随业务系统演进和升级，需要在消息中间件上添加一个消息服务层。消息服务层将底层消息中间件封装起来，隐藏消息中间件的具体操作细节，对上层提供统一风格的协议转换、消息路由和服务端点等功能，从而使数据传输系统成为独立完整的服务。我们将这种具有一致管理界面的消息管理平台定义为消息服务层中间件。

Apache Camel，提供不同系统之间消息传递的模式，提供大量已实现不同协议的网关功能的组件；核心是一个路由引擎，可支持丰富灵活的路由规则，包括动态路由。

条件路由功能：

from("kafka:localhost:9092?topic=input_events2&groupId=CamelStaticRouteExample&autoOffsetReset=latest&serializerClass=kafka.serializer.StringEncoder")
.process(new Processor() {
	@Override
	public void process(Exchange exchange) throws Exception {
		System.out.println(String.format("get event[%s]", exchange.getIn().getBody(String.class)));
		exchange.getIn().setHeader(KafkaConstants.KEY, UUID.randomUUID().toString());
		exchange.getIn().setHeader("event_type", JSONObject.parseObject(exchange.getIn().getBody(String.class)).getString("event_type"));
		exchange.getIn().removeHeader(KafkaConstants.TOPIC);
		// 必须删除KafkaConstants.TOPIC，否则Camel会根据这个值无限循环发送
	}
})
.choice()
.when(header("event_type").isEqualTo("click"))
.to("kafka:localhost:9092?topic=click&requestRequiredAcks=-1")
.when(header("event_type").isEqualTo("activate"))
.to("kafka:localhost:9092?topic=activate&requestRequiredAcks=-1")
.otherwise()
.to("kafka:localhost:9092?topic=other&requestRequiredAcks=-1");

动态路由功能：

from("kafka:localhost:9092?topic=input_events2&groupId=CamelSwitchRouteExample&autoOffsetReset=latest&serializerClass=kafka.serializer.StringEncoder")
.process(new Processor() {
	@Override
	public void process(Exchange exchange) throws Exception {
		System.out.println(String.format("get event[%s]", exchange.getIn().getBody(String.class)));
		exchange.getIn().setHeader(KafkaConstants.KEY, UUID.randomUUID().toString());
		exchange.getIn().setHeader("event_type", JSONObject.parseObject(exchange.getIn().getBody(String.class)).getString("event_type"));
		// 必须删除KafkaConstants.TOPIC，否则Camel会根据这个值无限循环发送
		exchange.getIn().removeHeader(KafkaConstants.TOPIC);
	}
})
.dynamicRouter(method(DynamicRouter.class, "slip"));

@Slf4j
public class DynamicRouter {

	public String slip(Exchange exchange) {
		String eventType = exchange.getIn().getHeader("event_type", String.class);
		if (StringUtils.isEmpty(eventType)) {
		    return null;
		}
		List<String> endpoints = getEndpoints(eventType);
		if (CollectionUtils.isEmpty(endpoints)) {
		    return null;
		}
		Integer index = exchange.getProperty("currentEndpointIndex", Integer.class);
		if (index == null) {
		    index = 0;
		}
		exchange.setProperty("currentEndpointIndex", index + 1);
		if (index >= endpoints.size()) {
		    return null;
		}
		String endpoint = endpoints.get(index);
		String topic = parseTopicFromEndpoint(endpoint);
		exchange.getIn().setHeader(KafkaConstants.TOPIC, topic);
		log.info("send event[%s] to endpoint[%s]", exchange.getProperty("eventId"), endpoint);
		return endpoint;
	}
	
	private List<String> getEndpoints(String eventType) {
		Map<String, List<String>> points = new HashMap<>();
		points.put("click", Arrays.asList(
		        "kafka:localhost:9092?topic=subsystem1&requestRequiredAcks=-1",
		        "kafka:localhost:9092?topic=subsystem2&requestRequiredAcks=-1"));
		points.put("activate", Arrays.asList(
		        "kafka:localhost:9092?topic=subsystem1&requestRequiredAcks=-1",
		        "kafka:localhost:9092?topic=subsystem2&requestRequiredAcks=-1",
		        "kafka:localhost:9092?topic=subsystem3&requestRequiredAc-ks=-1"));
		points.put("other", Arrays.asList(
		        "kafka:localhost:9092?topic=subsystem2&requestRequiredAcks=-1",
		        "kafka:localhost:9092?topic=subsystem3&requestRequiredAc-ks=-1"));
		return points.get(eventType);
	}
	
	private String parseTopicFromEndpoint(String endpoint) {
		String[] params = endpoint.split("\\?")[1].split("&");
		for (String param : params) {
		    String[] splits = param.split("=");
		    if ("topic".equals(splits[0])) {
		        return splits[1];
		    }
		}
		return null;
	}
}

数据存储

存储的设计原则

在实时流计算系统中，会遇到以下5种涉及数据存储的场景。

1. 实时流计算：经常需要记录各种状态，通常是各种聚类数据和历史信息数据。这些状态种类多样，且需要频繁地更新和读取，对数据的灵活性和访问性能有着较高的要求。实时流计算中的状态存储，通常选择NoSQL数据库，这有两方面的原因。一方面，NoSQL数据库的数据模型更加灵活，可简化程序逻辑。例如，Redis支持丰富的数据结构，让我们在计算和维护各种状态时非常便捷。另一方面，NoSQL数据库的性能通常比传统关系型数据库的性能更好。例如，Redis单节点能支持每秒10万次的读写性能，非常适合记录实时流计算中的各种状态。
2. 离线分析：经过实时流计算后，数据本身和计算结果通常会被保存下来，一方面可以做数据备份，另一方面可以做各种报表统计或离线分析。针对这种目的的数据存储，通常使用诸如HDFS的大数据文件系统。Hadoop已经非常成熟，构建在其上的查询和分析工具也多种多样，如Hive、HBase和Spark等。这些分析工具统一在Hadoop的生态体系内，给以后更多的方案探索和选择留有很大余地。
3. 点查询：除了数据备份、离线报表或离线分析这类数据存储外，还有一类偏重于提供实时查询计算结果的存储。这种查询更多是一种点查询，即根据一个或多个键来查询相应的值。针对这种目的的查询，通常选择NoSQL数据库并结合缓存的存储方案。当然，如果要查询的结果是结构化数据，则也可以使用关系型数据库。不过对于这种目的的查询而言，通常需要较高的请求处理能力和较低的时延，因此应该尽量避免复杂查询，如join操作。由于文档数据库采用JSON格式组织和存储数据，可以灵活地设计数据结构且避免不必要的关联计算，因此文档数据库非常适用于点查询。
4. Ad-Hoc查询：除了点查询外，还有一类用于交互式查询的存储方案。例如，通常实时流计算的计算结果会通过UI来呈现，如果UI要提供交互式的查询体验，那么这会涉及Ad-Hoc查询。对于这类查询的存储方案选择，在设计时一定要考虑到前端UI的需求变化，而不能选择一个“僵硬”的数据存储方案；否则当UI需求发生变化，需要调整各种查询条件时，这对后端存储的变更可能是一个巨大且痛苦的挑战。针对这种情况，推荐使用搜索引擎一类的存储方案，如ES。
5. 关系型数据库：最传统的关系型数据库技术及其结构化查询语言几乎在任何的数据系统中都不会缺席。所以，关系型数据库在实时流计算系统中也有一席之地。特别是在数据量不大，变更不太频繁时，关系型数据库相比NoSQL数据库具有非常明显的优势。一方面，成熟的关系型数据库技术十分适用于那些对数据的完整性和一致性要求非常严格的场景。另一方面，标准的结构化查询语言是大多数开发、数据和运维人员熟悉的查询工具，使用起来更加方便。在实时流计算系统中，典型的关系型数据库使用场景包括存储各种元数据和业务实体数据。

点查询

点查询，指通过给定主键或索引查询数据库中某条记录，每次最多返回一条记录。

数据灵活性：当数据的结构或字段变更时，是否需要大幅度地修改原有数据库表定义。关系型数据库不适合。

MongoDB类的文档数据库的优点：

• 文档数据天然与面相对象概念相对应，可省略ORM层；
• 查询更加直接、高效，支持文档（对象）嵌套；
• 文档数据库的数据结构更加灵活，方便加字段。

MongoDB集群注意事项：

• 必须合理指定Shard Server的内存；
• 最好为集合指定一个用于分区的shard key：只有分片集合才能被分布在多个节点上，集群才具备水平扩展能力；
• 最好为数据设置过期淘汰时间：没有数据是永远有用的；
• 为查询设置合适的索引；
• 单个集合的数据量不能过大。

严格控制MongoDB每个集合的数据量：

• 可保证在这个集合上的查询都可即时返回，不会出现一个查询堵死其他所有查询的情况；
• 数据的有效性本身就是有时间范围的；
• 某些对数据的查询本身就具有时间上的局部连续性。

使用按时间分表的方式来实现数据过期淘汰的优点：

• 每张表的数据量完全可控，维持在一个较小的范围内，可以稳定地控制每次请求响应的速度；
• 每张表代表一个时间段的数据，当出了问题需要复盘，或者需要查找问题时，非常容易定位问题数据；
• 当数据结构变更时，可以明确地限定变更时间。因为旧时间段的数据和新时间段的数据是完全隔离开的，不会出现冲突的问题；可以逐表地迁移数据，而不会出现一个遍历查询卡住其他所有线上查询的情况。

缺点：

• 如果不知道点查询对象的时间段，就需要依次查询所有表。这一点可以通过在查询时带上一个时间戳参数的方案来避免。通过这个时间戳，可以提示（hint）要查询的数据大概在哪个时间段，从而缩小查找范围。
• 如果查询的目标在多个时间段内，就需要对多次查询结果再做一次合并。这个问题不会出现在点查询这种情形下，因为点查询最多只返回一个结果。但是在问题复盘或分析问题时，如果需要做这种跨表查询，就需要分析人员自行写脚本来分析了。

Ad-Hoc查询

即席查询，是用户为了某个查询目的，灵活选择查询条件并提交数据库执行，最终生成相应查询结果的过程。

报表系统：需求灵活多变、查询千变万化、需要近实时返回结果、查询的频次低。

倒排索引适合报表系统。

ES四个特点非常适用于报表系统：

• 支持丰富的过滤、分组、聚合、排序等查询，可灵活地从一或多个维度分析数据；
• OLAP性能十分优异。在TB级别规模下，也可实现各种秒级别准实时OLAP查询；
• ES集群搭建和扩展非常容易、稳定、可靠；
• 数据存入ES，不需要专门预先针对OLAP查询设计各种聚合任务。使用ES这类数据存储和查询都非常灵活的存储方案可以减少太多以后的麻烦。

分索引的方式可分为3种：

1. 按时间分片：按时间分片是指根据时间周期，在每个新的时间周期使用一个新的索引，如按天分片、按小时分片等。按时间分片的好处在于实现简单、数据时间范围清晰明确、容易实现TTL。但是如果时间周期不好选择，或者数据流量在每个周期的变化比较大，就会造成每个索引内数据量的分布不均匀，索引数过多或者过少，从而给运维带来麻烦。
2. 按数据量分片：按数据量分片是指根据索引内记录的条数来决定是否使用新的索引。例如，如果平均每条记录是1KB，每个索引存放两千万条记录，那么，当索引中记录的数量超过两千万条时，就创建一个新的索引来存放新的数据。按数据量分片的好处在于每个索引数据量比较均匀。如果非要说缺点的话，这种方式的缺点就是不能通过索引名直接确定里面数据的时间范围
3. 同时按时间和数据量分片：既可保留每个索引内数据比较均匀的优点，还可通过索引名直接确定里面数据的时间范围。但是在代码实现时相对更复杂些。

离线分析

离线系统的作用包括：

• 数据存储和ETL处理；
• 离线数据分析和模型训练；
• 离线报表统计。

离线任务的3个方面：存储、处理和分析、调度。

存储

Flume常用于从Kafka拉取数据写入HDFS。Flume使用注意事项：

• 小文件：Flume将数据写入HDFS时可以设置3种滚动条件，即按时间间隔滚动(rollInterval)、按文件大小滚动(rollSize)和按事件数滚动(rollCount)。小文件合并。
• 时间戳：Flume使用事件头部的timestamp字段作为分区时间依据。如果需要使用事件发生的时间而不是Flume接收到事件的时间作为分区时间依据，则需特殊处理。
• HDFS高可用

Apache Camel，可统一且方便地管理数据在不同端点之间的传递，这部分解决数据入库任务的管理问题。但对这种任务管理的支持还不是一步到位的，依旧需要自己开发诸如集群化、监控、管理和UI之类的功能。

Apache NiFi，大数据集成平台，优良特性：

• 通过图形化界面创建、管理、监控各种ETL任务，使用直观方便；
• 集群化的运行环境一方面能够集中管理各种ETL的任务，不需要像Flume或Camel那样管理零散运行实例，另一方面能够更加一致地对集群处理能力进行水平扩展；
• 简单且独立于其他如YARN或Mesos等资源管理框架的集群方案，让其具有更少的依赖，部署、管理和维护起来非常方便。

类似平台：Apache Gobblin。

处理和分析

Hive：在将数据与表绑定起来时，应尽量使用外部表。只有在需要创建和使用临时表时，才使用内部表。临时表用完后要删除。

Spark：RDD（Resilient Distributed Datasets，分布式弹性数据集）和DataFrame两个概念都是对数据的矩阵表示。

调度

Azkaban：解决调度作业之间的相互依赖问题。需严格控制被调度任务的占用内存。内存过大，调度任务会被操作系统随机杀死。

关系型数据库查询

关系型数据库主要用于存储元数据和业务实体数据。

数据高可用，MySQL5.7引入Group Replication，基于Paxos实现，支持两种模式，即单主模式和多主模式。

NewSQL=SQL+NoSQL，大数据场景下关系型数据库，代表：TiDB和CockroachDB。

小结

注意点：

• 根据计算类型选择最合适的存储方案；
• 在实时流计算系统中，没有数据是永远有效的，必须设置超期时间；
• 在设计之初合理预估将来的数据量规模，规划好集群规模并制订扩展计划；
• 单表过大容易变成灾难，必要时对数据按时间分表存储。

服务治理和配置管理

服务治理

流服务和微服务

流服务：当一个服务模块的输入和输出都是流时，好处在于其可以直观地描述业务执行流程。使用DAG来描述执行流程，DAG每个节点代表一个业务单元，每个业务单元负责一定的业务逻辑。

Spring Cloud

一笔带过：

• 服务注册及发现
• 负载均衡
• 容错保护
• 配置管理
• 链路追踪
• 服务网关

面向配置编程

一般情况下，配置是程序的附属资源；但风控类系统，配置化信息（特征、规则、模型、决策）比程序更加核心。

面向配置编程思想

脚本即配置，配置即脚本。规则引擎Drools使用的DRL文件，是配置也是脚本。

面向配置编程：配置比程序更重要，更核心。

面向配置编程包含两个部分：

• 配置：当涉及业务逻辑时，配置才是描述系统执行逻辑的核心所在。因此，针对具体业务场景设计合适的配置项目和配置组织结构，是配置设计的核心所在；
• 引擎：引擎的开发应该围绕着配置来进行。当配置设计好后，应按照配置表达的业务逻辑，开发对应的执行引擎。

面向配置编程的好处：

• 灵活和轻便：当面向配置编程的引擎在开发完成后，只要整体逻辑不变，调整业务只需要编写或修改对应的配置文件就可以了，而不需要再修改程序并重新构建测试和部署上线，极大地缩短业务上线周期；
• 简洁和透明。在使用面向配置编程时，编写配置文件相比程序开发简单和透明很多，因为配置编写的过程直接是实现业务逻辑的过程；
• 抽象和泛化。面向配置编程开发的引擎相当于一个脚本解释器，它是对业务执行流程的终极抽象。配置的灵活性，使得只要是在这个业务执行流程的框架下，我们可以任意地设置业务流程的各种指标或参数，可以说是对业务执行流程的终极泛化。

更高级的配置：领域特定语言

DSL：Domain Specified Languag，领域特定语言。相比于通用语言，DSL表达能力有限，不是图灵完备（不能用DSL实现C语言，但是能够用C语言实现DSL）。

动态配置

配置分类：

• 静态配置：程序启动后无需修改，一般会放在配置文件；
• 动态配置：程序启动后需要修改，并能实现应用无需重新部署而重新生效，一般会放在配置中心；

动态配置的复杂性：

• 分布式系统环境
• 安全性
• 版本控制
• 监控

动态配置的实现方式：

• 控制流：通过控制流与数据流的关联（union或join）操作，就可以将控制信息作用到数据流上。

• 共享存储：将配置存放在共享数据库中，当配置发生变更时，先将配置写入共享数据库，然后通过配置使用方轮询或者通知配置使用者配置变更的方式，配置使用者即可重新读取更新后的配置。

• 配置服务：单独抽取出一个微服务，使用Spring Cloud Config + Spring Cloud Bus，具体的MQ中间件一般会选择RabbitMQ。

将前端配置与后端服务配置隔离开

分离

应用案例

实时流数据特征提取引擎

DSL包含7个主要概念：

• 输入流：source，定义事件的输入流。实时流计算特征提取引擎以Kafka等消息中间件作为事件的输入流；
• 输出流：sink，定义事件的输出流。当特征提取引擎对事件提取完特征后，将特征附加（append）到事件上，再将附加了特征的事件输出到Kafka等消息中间中；
• 字段：filed，通过字段映射功能来设定特征引擎感兴趣的字段与原始事件字段之间的对应关系；
• 算子：operator，OPERATOR(window, event_type, target[=value], on1[=value], on2[=value], ...)，例子

# 过去一周内在同一个设备上交易次数
COUNT(7d, transaction, device_id)
# 过去一周内在设备`d000001`上交易次数
COUNT(7d, transaction, device_id=d000001)
# 过去一天同一用户的总交易金额
SUM(1d, transaction, amount, userid)
# 过去一天用户"ud000001"的总交易金额
SUM(1d, transaction, amount, userid=ud000001)
# 过去一周内在同一个设备上注册的用户登录过的设备数
FLAT_COUNT_DISTINCT(7d, login, device_id, SET(7d, create_account, userid, device_id))
# 过去一周内在设备`d000001`上注册的用户登录过的设备数
FLAT_COUNT_DISTINCT(7d, login, device_id, SET(7d, create_account, userid, device_id=d000001))

• 函数：function，F_FUNCTION(on1[=value], on2[=value], ...)
• 宏函数：macro
• 操作模式：mode，3种计算模式：update、get和upget。

特征引擎包含3层：DSL解析层、执行计划执行层和状态存储层。

使用Flink实现风控引擎

如下图