创建和管理数据流

从用户界面创建数据流

从左侧的导航窗格中，选择数据流以打开“数据流”页。
单击添加数据流按钮。此时将显示 "Add Data Stream" diaglox 框。
在数据流信息页上，在名称框中输入唯一的资源名称。

名称必须介于 1 到 32 个字符之间。以字母开头。使用稳定名称，因为检查点将限定于流合同。单击下一步。
使用业务所有者、源线索、表、使用者和恢复目标添加说明。
在源选项页上，为下图中显示的选项提供值：
- 线索名称：源线索文件的名称。线索名称必须是标识服务接受的源线索前缀或源端点的两个字符。
- 线索子目录：存储线索文件的子目录的路径。这是在源线索所在的 GoldenGate 部署下存储的目录（当该目录不在默认位置时）。
- 编码：此选项控制数据流的记录编码。目前，支持 JSON 编码。将 YAML 编辑器或 REST 配置用于高级 JSON 选项。
- 缓冲区大小：此值控制数据流服务中使用的内存缓冲区大小。数据流处理服务将在内存中消息队列的总字节大小超过指定值后刷新该队列，并将记录传送给客户机。范围 131,072 到 10,485,760；默认值 1,048,576。这是不持久的保留。
- 服务质量：此选项定义数据流重复的禁止和恢复行为。exactlyOnce、atLeastOnce 或 atMostOnce。
  
  在选择非默认值之前，请参见 Checkpointing，Recovery，and Delivery Semantics 。
  
  数据流服务支持三个级别的服务质量：
  - 仅限于一次：此模式是限制性最强的模式，在恢复在源跟踪中查看 RESTART OK/ABEND 记录的过程中，服务会筛选出任何重复记录。客户不会看到重复的记录。如果服务无法找到具有给定上次处理位置的记录，则会出现错误。
  - 至少一次：此模式不会在恢复期间隐藏重复记录，也不会在源线索中看到 RESTART OK/ABEND 记录。客户机可能会在数据流中看到重复的记录。此外，如果服务无法找到具有给定上次处理位置的记录，则会发生错误。
  - 最多一次：此模式在恢复期间隐藏重复记录或在源线索中看到 RESTART OK/ABEND 记录。客户机在数据流中看不到重复的记录。如果服务无法找到具有给定上次处理位置的记录，它将查找下一个可用记录并前进。
- CloudEvents 格式：数据流处理服务支持以 CloudEvents 格式传输数据记录。它将 GoldenGate 记录包装在 CloudEvents JSON 中。默认情况下处于禁用状态。默认情况下，此格式处于禁用状态，可以在创建数据流通道时通过使用切换开关启用属性来控制。
在筛选选项页上，包含和排除筛选规则的选项可用，如下图中所示：

指定筛选规则选项，如下所示：
- 规则操作：选择“排除”或“包括”选项。
- 筛选器类型：筛选器类型包括以下内容：
  
  对象类型：可以从下拉列表中选择多个选项，包括 DML 、 DDL 、 INSERT 、 UPDATE 、 UPSERT 和 DELETE 。
  
  对象名称：以前创建的过滤对象的名称。

单击添加将筛选规则添加到数据流流程中。

单击创建数据流。您将返回到列出数据流的分发服务主页。
单击创建的数据流以查看该流的 YAML 文档数据流 AsyncAPI 定义。

从用户界面编辑数据流配置

要编辑数据流配置，请执行以下操作：

在左侧导航窗格中，单击“Data Streams（数据流）”。
选择需要修改的数据流的名称。
在“数据流”页中，使用操作列查看数据流详细信息、删除数据流并更改其筛选。
如果从操作菜单中单击更改筛选选项，则会显示编辑数据流筛选规则对话框。在此框中，您可以更改规则操作（包括、排除）和筛选器类型（对象名称或对象类型）。
单击添加以应用筛选。
单击提交以返回“数据流”页。

通过从名称列中单击数据流，可以进一步编辑数据流。这将显示完整的数据流配置。使用每个配置设置旁边的铅笔图标进行更改。您可以更改数据流使用的源线索文件、过滤规则和服务质量。您还可以使用 YAML 编辑器更改数据流配置，并使用 YAML 编辑器旁边的上载更改图标上载更改。

使用 REST API 创建数据流

将 POST /services/v2/stream/{streamName} 与管理员帐户一起使用。最小请求标识源线索：

{
  "$schema": "ogg:dataStream",
  "source": {
    "trail": "ea",
    "path": "./dirdat"
  }
}

生产请求通常会显式设置服务质量、筛选器、JSON 方案排放、元数据字段和大小控制。支持文件 data_stream_create_request.json 包含一个完整的已清理示例。

{
  "$schema": "ogg:dataStream",
  "description": "Streams sales order changes",
  "source": {"trail": "ea", "path": "./dirdat"},
  "qualityOfService": "exactlyOnce",
  "bufferSize": 1048576,
  "tcpKeepAliveTimeout": 120,
  "cloudEventsFormat": false,
  "rules": [
    {
      "action": "include",
      "filter": {"objectNames": ["SALES.ORDERS", "SALES.ORDER_ITEMS"]}
    }
  ],
  "encoding": {
    "json": {
      "$schema": "ogg:outputFormatJson",
      "model": "op",
      "includeJSONSchemaRecord": true,
      "includeObjectMetadataRecords": true,
      "includeStreamMetadataRecord": true,
      "metaFields": {
        "qualifiedTable": true,
        "opType": true,
        "opTs": true,
        "pos": true,
        "xid": true,
        "csn": "csn",
        "opSeqno": "op_seqno",
        "transInd": "trans_ind"
      }
    }
  }
}

顶级配置属性

下表提供了用于配置流的配置属性。

属性	类型/默认值	说明
`$schema`	字符串 / `ogg:dataStream`	配置方案标识符。
`source`	对象/必需	源线索和可选路径。
`description`	字符串/无	描述数据流的 1 到 4,096 个字符。
`qualityOfService`	枚举 / `exactlyOnce`	`atLeastOnce`、`exactlyOnce` 或 `atMostOnce`。
`bufferSize`	整型 / 1,048,576	刷新阈值（字节）。最小值 131,072；最大值 10,485,760。
`tcpKeepAliveTimeout`	整数 / 120	TCP 保持活动超时（以秒为单位）。最小值 1；最大值 10,800。
`cloudEventsFormat`	布尔值 / `false`	启用 CloudEvents JSON 信封。
`rules`	数组/无	1 至 1,024 包括或排除规则。
`encoding`	字符串或对象 / `json`	将 `"json"` 用于默认值，将 `{"json": {...}}` 用于高级输出设置。

检索、更新、列出和删除数据流

使用下表可以管理或删除数据流：

任务	方法和途径	角色
列出流	`GET /services/v2/stream`	用户
获取配置	`GET /services/v2/stream/{streamName}`	运算符
获取生成的 YAML	`GET /services/v2/stream/{streamName}/yaml`	用户
更新配置	`PATCH /services/v2/stream/{streamName}`	管理员
更新 YAML	`PATCH /services/v2/stream/{streamName}/yaml`	管理员
删除流	`DELETE /services/v2/stream/{streamName}`	管理员

注意：更改字段名称、操作字符串、记录模型、筛选器或模式排放会更改使用者合同。在生产部署之前，从现有检查点协调更改、对合同进行版本控制和测试重新启动。

使用生成的 YAML 作为已部署的合同

生成的 YAML 包括特定于部署的服务器、通道地址、begin 查询参数、消息有效负载和支持的记录方案。在创建后和每次配置更改后检索它。将审核副本与使用者版本一起存储，但在部署验证期间始终将其与运行时版本进行比较。