Pravega 与 Flink 的设计理念类似，都以流为基础实现流批统一的接口便于应用使用。Pravega 团队也希望与 Flink 一起打造从底层存储到上层计算的统一大数据流水线架构。在开发层面，Pravega 与 Flink 也有着深度的合作，Pravega Flink Connector 的开发，特别是在 Flink 的端到端的仅一次语义实现的过程中，都得到了 Apache Flink PMC 成员的通力协作和大力支持。本文将从API的简介及使用入手，重点介绍Pravega+Flink的流计算编程，

简介

Pravega是根据Apache 2.0许可证开源的流存储引擎，为连续流数据提供统一的Stream抽象。Pravega提供了持久化、强一致性以及高性能低延迟的数据存储。同样在实时大数据领域，Apache Flink是由Apache软件基金会开发的开源分布式处理引擎，用于对无界和有界数据流进行有状态的计算。 Flink提供高吞吐量、低延迟的流数据计算，以及对事件发生时间处理和状态管理的支持。Flink的应用程序在发生机器故障时具有容错能力，并且支持exactly-once语义。

Pravega与Flink的设计理念类似，都以流为基础实现流批统一的接口便于应用使用。Pravega团队也希望与Flink一起打造从底层存储到上层计算的统一大数据流水线架构。Pravega从诞生之初就积极参加Flink社区的活动，自从2017年起在每一次的Flink Forward大会上都有相关内容的分享，包括在2018年12月第一次在中国举办的Flink Forward同样也有Pravega中国团队的参与。在开发层面，Pravega与Flink也有着深度的合作，Pravega Flink Connector 的开发，特别是在Flink的端到端的仅一次语义实现的过程中，都得到了Apache Flink PMC成员的通力协作和大力支持。

Pravega Flink Connectors

Pravega Flink Connectors实现了Flink的接口，提供了对Pravega Stream的读取和写入，并且结合Flink的Checkpoint机制提供了端到端的exactly-once处理语义（详情可见上一篇文章）。Flink对数据有两种读和写处理办法，对应的，每一种API都需要定义Flink程序的数据源(Source)和数据汇(Sink)。Pravega Flink Connector打通存储与计算之间的通道，Pravega就可以作为统一的流存储和消息总线，用户可以使用统一的Flink API在Pravega Stream上进行批或者流计算，构建一个完整的实时数据仓库。

Flink在数据源上自底向上有着4层抽象的API，对于Pravega而言需要提供三种不同的API来满足不同的使用需求。

DataStream API
DataSet API
Table API

API抽象

在介绍这三种API之前，我们首先先了解一下这些API所共有的参数。

共有配置

1. PravegaConfig 类

Pravega Flink Connectors提供了一个配置对象PravegaConfig，用于为Flink配置Pravega的上下文。PravegaConfig会自动从环境变量、配置文件属性和程序运行时参数进行配置。

PravegaConfig信息来源如下：

设置	环境变量/ 系统属性 / 程序参数	缺省值
Controller URI	`PRAVEGA_CONTROLLER_URI` / `pravega.controller.uri` / `--controller`	`tcp://localhost:9090`
Default Scope	`PRAVEGA_SCOPE` / `pravega.scope` / `--scope`	-
Credentials	-	-
Hostname Validation	-	`true`

创建PravegaConfig

创建PravegaConfig实例的推荐方法是利用Flink的ParameterTool

ParameterTool params = ParameterTool.fromArgs(args);
PravegaConfig config = PravegaConfig.fromParams(params);

如果您的应用程序不使用Flink提供的ParameterTool类，也可以使用fromDefaults创建PravegaConfig：

PravegaConfig config = PravegaConfig.fromDefaults();

此外，PravegaConfig也提供了 builder-style API允许用户覆盖默认配置：

PravegaConfig config = PravegaConfig.fromDefaults()
    .withControllerURI("tcp://...")
    .withDefaultScope("SCOPE-NAME")
    .withCredentials(credentials)
    .withHostnameValidation(false);

使用PravegaConfig

随连接器库提供的所有各种源和接收器类都具有builder-style API，可接受通用配置的PravegaConfig。通过withPravegaConfig将PravegaConfig对象传递给相应的构建器。例如：

PravegaConfig config = ...;

FlinkPravegaReader<MyClass> pravegaSource = FlinkPravegaReader.<MyClass>builder()
    .forStream(...)
    .withPravegaConfig(config)
    .build();

值得注意的是，source或sink的stream可以使用完整的stream名称（使用/分隔scope和stream，例如my-scope/my-stream），也可以在设置DefaultScope的情况下使用不完整的stream名称（例如my-stream）。

2. 序列化/反序列化

序列化/反序列化是指Flink程序中的数据元素与Pravega Stream中存储的二进制消息相互转换的过程。

Flink定义了数据序列化/反序列化的标准接口，核心接口是：

Flink内置的序列化器包括：

Pravega Connector可以使用Flink的序列化接口。例如，要将每个事件读取为UTF-8字符串：

DeserializationSchema<String> schema = new SimpleStringSchema();
FlinkPravegaReader<String> reader = new FlinkPravegaReader<>(..., schema);
DataStream<MyEvent> stream = env.addSource(reader);

与其他应用程序的互操作性

更常见的情况是，Pravega的数据由其它客户端程序注入，使用Flink处理。此类应用程序使用的Pravega客户端库定义了用于处理事件数据的io.pravega.client.stream.Serializer接口。Pravega提供了内置的适配器，实现了序列化方法的转化，使得Flink程序能够正常读写Pravega中的数据

下面是一个示例，将实现了Pravega Serializer 接口的内置Java POJO类序列化器JavaSerializer传递给适配器的构造函数，最终数据转化为DataStream<MyEvent>进行进一步处理：

import io.pravega.client.stream.impl.JavaSerializer;
...
DeserializationSchema<MyEvent> adapter = new PravegaDeserializationSchema<>(
    MyEvent.class, new JavaSerializer<MyEvent>());
FlinkPravegaReader<MyEvent> reader = new FlinkPravegaReader<>(..., adapter);
DataStream<MyEvent> stream = env.addSource(reader);

Pravega序列化程序必须实现java.io.Serializable才能在Flink程序中使用。

3. Stream Cuts

StreamCut表示Pravega流中的特定位置，可以从与Pravega客户端的各种API交互中获得，它包含一组segment和offset的键值对。偏移量始终指向事件边界，因此没有指向不完整事件的offset。Pravega中的Checkpoint底层也由这一API实现。读客户端可以接受StreamCut作为给定流的开始和/或结束位置。由于数据在不断产生以及不断地下沉至第二级存储，stream的开始和结束位置都会发生变化，因此Pravega使用StreamCut.UNBOUNDED表示Stream中的不断变化的位置。这样的设计有助于使得Flink使用一套统一的API读取用户自定义的有边界和无边界的数据流。

DataStream API

DataStream API是Flink最常用的API，主要负责数据流的处理。数据流通过Source来创建，结果通过Sink返回，中间可以进行丰富的有状态的transformation操作。Pravega Flink Connectors扩展了Flink的RichParallelSourceFunction和RichSinkFunction，以DataStream API实现了Pravega的数据读写。

FlinkPravegaReader

使用io.pravega.connectors.flink.FlinkPravegaReader的实例作为Flink程序的数据源(Source)。FlinkPravegaReader读取Pravega的一个或多个Stream，抽象为Flink的DataStream。

使用StreamExecutionEnvironment::addSource方法将Pravega Stream作为DataStream打开。

代码示例:

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// Define the Pravega configuration
PravegaConfig config = PravegaConfig.fromParams(params);

// Define the event deserializer
DeserializationSchema<MyClass> deserializer = ...

// Define the data stream
FlinkPravegaReader<MyClass> pravegaSource = FlinkPravegaReader.<MyClass>builder()
    .forStream(...)
    .withPravegaConfig(config)
    .withDeserializationSchema(deserializer)
    .build();
DataStream<MyClass> stream = env.addSource(pravegaSource);

Pravega Flink Connectors提供builder-style API构造FlinkPravegaReader的实例，通常需要指定PravegaConfig，读取的Stream名称以及反序列化方法。如果需要并行读取多个Stream，可以反复调用forStream。 Stream也可以跨scope被Flink所读取，只需指定完整的stream名称scope/stream 即可。更多参数可参阅此表。forStream提供了一个重载方法，可以接受StreamCut类型的参数以处理历史流数据。

FlinkPravegaReader支持并行化，可使用setParallelism方法配置要执行的并行实例的数量。每个实例消耗一个或多个segment。

读客户端与Flink checkpoints和savepoints兼容，其中会包含从正确位置恢复所需的所有信息，读客户端支持倒回到Pravega Checkpoint位置从故障中恢复。

Checkpoint的工作过程分为两步：

FlinkPravegaReader在初始化期间注册 ReaderCheckpointHook，Job manager中的master hook处理程序启动triggerCheckpoint request 到ReaderCheckpointHook。 ReaderCheckpointHook处理程序通知Pravega检查当前读客户端状态。这是一个非阻塞调用，一旦Pravega读者完成了检查点，就会返回。
Pravega将发送CheckPoint事件作为数据流流的一部分，并且在接收事件时，FlinkPravegaReader将启动triggerCheckpoint请求以有效地让Flink继续并完成检查点过程。

FlinkPravegaReader默认开启性能指标(Metrics)监控，可使用.enableMetrics（false）选项禁用。用户可以实时地观察Flink消费Pravega数据的运行状态，性能指标包括：

名称	说明
`readerGroupName`	Reader组名称
`scope`	Reader组的范围名称
`streams`	作为Reader组一部分的流的完全限定名称 (i.e., `scope/stream`)
`onlineReaders`	当前在线/可用的Reader
`segmentPositions`	`StreamCut`信息，指示Reader到目前为止所阅读的位置。
`unreadBytes`	尚未读取的总字节数

FlinkPravegaWriter

使用io.pravega.connectors.flink.FlinkPravegaWriter的实例作为Flink程序里面的数据汇(Sink)。使用DataStream::addSink方法将写客户端的实例添加到Flink程序中。

代码示例：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// Define the Pravega configuration
PravegaConfig config = PravegaConfig.fromParams(params);

// Define the event serializer
SerializationSchema<MyClass> serializer = ...

// Define the event router for selecting the Routing Key
PravegaEventRouter<MyClass> router = ...

// Define the sink function
FlinkPravegaWriter<MyClass> pravegaSink = FlinkPravegaWriter.<MyClass>builder()
   .forStream(...)
   .withPravegaConfig(config)
   .withSerializationSchema(serializer)
   .withEventRouter(router)
   .withWriterMode(EXACTLY_ONCE)
   .build();

DataStream<MyClass> stream = ...
stream.addSink(pravegaSink);

FlinkPravegaWriter同样利用builder API构造，通常需要指定PravegaConfig，写入的Stream名称，写入的模式，路由函数以及序列化方法。完整参数列表可参阅此表。

写入Pravega stream的每个事件都有一个关联的路由键。路由键是事件分segment的依据以及有序性保证的基础，详细信息可参阅之前Pravega系列文章。用户需要提供io.pravega.connectors.flink.PravegaEventRouter接口的实现，例如，为了保证特定于传感器 id 的写入顺序，您可以提供如下所示的路由实现。

private static class SensorEventRouter<SensorEvent> implements PravegaEventRouter<SensorEvent> {
        @Override
        public String getRoutingKey(SensorEvent event) {
            return event.getId();
        }
    }

用户可以进一步使用FlinkPravegaUtils::writeToPravegaInEventTimeOrder 方法将给定的 DataStream按照事件时间顺序写入Pravega流，该方法将自动对事件按事件时间进行排序 (基于每个键)。

FlinkPravegaWriter根据性能以及持久化保证的权衡，支持三种写模式:

最多一次（BEST_EFFORT）：任何写入失败都将被忽略, 因此可能会出现数据丢失。
最少一次（ATLEAST_ONCE）：所有事件都在Pravega持续存在。由于重试或在失败和后续恢复的情况下, 可能会发生重复的事件。
仅一次（EXACTLY_ONCE）：集成Flink Checkpoint功能使用事务性写入，在 Pravega 中保留所有事件且仅写入一次。

默认情况下, 启用ATLEAST_ONCE选项。

DataSet API

DataSet API是Flink进行批处理程序中使用的API。Pravega Stream作为数据源和数据汇。Pravega Flink Connectors扩展了Flink的RichInputFormat和RichOutputFormat，实现了DataSet API的Pravega读写。参数与DataStream API类似。

FlinkPravegaInputFormat

使用io.pravega.connectors.flink.FlinkPravegaInputFormat的实例用作Flink批处理程序中的数据源。输入格式将流的事件作为DataSet（Flink Batch API的基本抽象）读取。此输入格式并行处理stream segments，而不遵循路由键顺序。使用ExecutionEnvironment::createInput方法将Pravega Stream作为DataSet打开。

代码示例：

// Define the Pravega configuration
PravegaConfig config = PravegaConfig.fromParams(params);

// Define the event deserializer
DeserializationSchema<EventType> deserializer = ...

// Define the input format based on a Pravega stream
FlinkPravegaInputFormat<EventType> inputFormat = FlinkPravegaInputFormat.<EventType>builder()
    .forStream(...)
    .withPravegaConfig(config)
    .withDeserializationSchema(deserializer)
    .build();

DataSource<EventType> dataSet = env.createInput(inputFormat, TypeInformation.of(EventType.class)
                                   .setParallelism(2);

FlinkPravegaOutputFormat

使用io.pravega.connectors.flink.FlinkPravegaOutputFormat的实例用作Flink批处理程序中的数据汇。使用DataSet::output方法将写客户端的实例添加到Flink程序中。

代码示例：

// Define the Pravega configuration
PravegaConfig config = PravegaConfig.fromParams(params);

// Define the event serializer
SerializationSchema<EventType> serializer = ...

// Define the event router for selecting the Routing Key
PravegaEventRouter<EventType> router = ...

// Define the input format based on a Pravega Stream
FlinkPravegaOutputFormat<EventType> outputFormat = FlinkPravegaOutputFormat.<EventType>builder()
    .forStream(...)
    .withPravegaConfig(config)
    .withSerializationSchema(serializer)
    .withEventRouter(router)
    .build();

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
Collection<EventType> inputData = Arrays.asList(...);
env.fromCollection(inputData)
   .output(outputFormat);
env.execute("...");

Table API

Table API 是一种关系型API，用户可以使用简单的SQL操作数据，降低了大数据分析的门槛。更重要的是，Table API统一了Flink里批和流不同的API。在同一套API下，批处理的查询返回有限数据集，而流处理的查询则会持续返回流式结果。因此，这也是Flink社区贡献的热点和Flink未来的发展重点。Pravega Flink Connectors提供了FlinkPravegaTableSource和FlinkPravegaTableSink使用Flink Table API读写Pravega数据，以便用户使用SQL语句操作数据。用户既可以在代码中使用Pravega Descriptor指定数据源，也可以通过声明式的YAML配置文件启用SQL client.

FlinkPravegaTableSource/Sink

Pravega Stream可以用作Flink Table程序中的Table源。 Flink Table API面向Flink的TableSchema类，它们描述了table字段。然后使用FlinkPravegaTableSource/Sink的具体子类将流数据解析为符合table模式的Row对象进行相应的读写。

FlinkPravegaTableSource 通过 TableEnvironment::registerTableSource连接，Sink的创建过程与Source类似，FlinkPravegaTableSink 通过 table.writeToSink(sink)写入。

代码示例：

以下示例使用Table API从Pravega Stream读取JSON格式的用户网站访问事件：

// define table schema definition
Schema schema = new Schema()
        .field("user", Types.STRING())
        .field("uri", Types.STRING())
        .field("accessTime", Types.SQL_TIMESTAMP()).rowtime(
                new Rowtime().timestampsFromField("accessTime")
                        .watermarksPeriodicBounded(30000L));

// define pravega reader configurations using Pravega descriptor
Pravega pravega = new Pravega();
pravega.tableSourceReaderBuilder()
        .withReaderGroupScope(stream.getScope())
        .forStream(stream)
        .withPravegaConfig(pravegaConfig);

 // Streaming Source
StreamExecutionEnvironment execEnvRead = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(execEnvRead);

StreamTableDescriptor desc = tableEnv.connect(pravega)
        .withFormat(new Json().failOnMissingField(true).deriveSchema())
        .withSchema(schema)
        .inAppendMode();

final Map<String, String> propertiesMap = DescriptorProperties.toJavaMap(desc);
final TableSource<?> source = TableFactoryService.find(StreamTableSourceFactory.class, propertiesMap)
        .createStreamTableSource(propertiesMap);

tableEnv.registerTableSource("MyTableRow", source);
String sqlQuery = "SELECT user, count(uri) from MyTableRow GROUP BY user";
Table result = tableEnv.sqlQuery(sqlQuery);
...

利用TableEnvironment::connect连接Pravega的相应stream，并通过withFormat方法指定数据格式，withSchema方法指定表结构。之后通过相应的工厂模式，以及传入的装饰器的Map来构造FlinkPravegaTableSource的具体子类。

FlinkPravegaTableSource/Sink支持Flink流和批处理环境。数据读取在流处理环境中使用FlinkPravegaReader/Writer实现；在批处理环境中使用FlinkPravegaInputFormat/OutputFormat实现。

快速入门

程序中使用

在编写Flink代码之前，需要确保Flink集群能够访问到Pravega集群，并且将Pravega Connector的依赖添加到项目中，pom.xml如下：

<dependency>
  <groupId>io.pravega</groupId>
  <artifactId>pravega-connectors-flink_2.11</artifactId>
  <version>0.3.2</version>
</dependency>

用户应该根据所运行的Pravega的版本使用对应的Pravega Connector版本。在运行/部署应用程序时，Pravega Flink Connector不属于Flink的核心运行时，因此用户需要保证其代码必须是应用程序代码artifacts （JAR文件）的一部分。

使用SQL客户端

Flink SQL Client是在Flink 1.6中引入的，旨在提供一种简单的方法来编写，调试和提交Table API程序到Flink集群，而无需编写Java或Scala代码。SQL Client CLI允许在命令行上检索和可视化运行的分布式应用程序的实时结果。

Pravega Stream支持通过Flink的SQL客户端使用标准SQL命令访问。为此，必须下载以下文件（可使用maven）并复制到Flink集群library 路径：$FLINK_HOME/lib

Pravega connector jar
Flink JSON jar（以json格式序列化/反序列化数据）
Flink Avro jar（以avro格式序列化/反序列化数据）

之后准备如下的SQL客户端YAML格式的配置文件，并确保任何相关的Pravega Stream已经创建。详细格式标准参见文档。

tables:
  - name: sample
    type: both
    update-mode: append
    # declare the external system to connect to
    connector:
      type: pravega
      version: "1"
      metrics: true
      connection-config:
        controller-uri: "tcp://localhost:9090"
        default-scope: wVamQsOSaCxvYiHQVhRl
      reader:
        stream-info:
          - stream: streamX
      writer:
        stream: streamX
        mode: atleast_once
        txn-lease-renewal-interval: 10000
        routingkey-field-name: category
    format:
      type: json
      fail-on-missing-field: true
      derive-schema: true
    schema:
      - name: category
        type: VARCHAR
      - name: value
        type: INT

functions: [] 

execution:
  # 'batch' or 'streaming' execution
  type: streaming
  # allow 'event-time' or only 'processing-time' in sources
  time-characteristic: event-time
  # interval in ms for emitting periodic watermarks
  periodic-watermarks-interval: 200
  # 'changelog' or 'table' presentation of results
  result-mode: table
  # parallelism of the program
  parallelism: 1
  # maximum parallelism
  max-parallelism: 128
  # minimum idle state retention in ms
  min-idle-state-retention: 0
  # maximum idle state retention in ms
  max-idle-state-retention: 0

deployment:
  # general cluster communication timeout in ms
  response-timeout: 5000
  # (optional) address from cluster to gateway
  gateway-address: ""
  # (optional) port from cluster to gateway
  gateway-port: 0

之后使用命令$FLINK-HOME/bin/sql-client.sh embedded -d <SQL_configuration_file> 以嵌入式模式运行SQL client shell，若能成功运行SELECT 'Hello World'，即可以运行SQL命令与Pravega进行交互。

Pravega系列文章计划

Pravega根据Apache 2.0许可证开源，0.5版本即将发布。我们欢迎对流式存储感兴趣的大咖们加入Pravega社区，与Pravega共同成长。本篇文章为Pravega系列的最后一篇，希望这系列文章能让你对流和流式实时计算有一个初步的了解，也欢迎感兴趣的朋友后续继续交流。下面是我们这个系列的文章标题，以备参考：

作者简介

滕昱：就职于 DellEMC 非结构化数据存储部门 (Unstructured Data Storage) 团队并担任软件开发总监。2007 年加入 DellEMC 以后一直专注于分布式存储领域。参加并领导了中国研发团队参与两代 DellEMC 对象存储产品的研发工作并取得商业上成功。从 2017 年开始，兼任 Streaming 存储和实时计算系统的设计开发与领导工作。

黄飞情，现就职于DellEMC，10年+ 存储、分布式虚拟化、云计算开发以及架构设计经验，现从事流存储和实时计算系统的设计与开发工作；

周煜敏，复旦大学计算机专业研究生，从本科起就参与 DellEMC 分布式对象存储的实习工作。现参与 Flink 相关领域研发工作。

参考链接

创作场景

如何利用开源 Flink 和 Pravega 搭建完整的流处理架构？