11 流

本节介绍在 Oracle AI Data Platform Workbench 中使用流数据或持续生成的数据。

主题：

• 关于流处理
• 使用工作流配置 Spark 结构化流处理

关于流处理

您可以使用 Apache Spark 结构化流处理功能，在 Oracle AI Data Platform Workbench 中近乎实时地处理流数据或持续生成的数据。

笔记本和工作流都支持 Apache Spark 结构化流处理。可以使用以下源和接收器从检查点位置读取流数据，将流数据写入到检查点位置。

表 11-1 支持的源和汇

源或接收器	支持？
卷路径 (/Volume/bronze/bucket1)	支持所有格式
工作区路径 (/Workspace/folder1/)	支持所有格式
目录中具有三个部分名称的表 (catalog.schema.table)	仅支持 Delta 格式 Parquet、CSV、JSON、ORC 格式不支持 示例 1：支持的代码 streaming_df = spark.readStream.format("delta").table('stdcatalog.stdschema.deltatable') streaming_df.writeStream.format("delta").outputMode("append").option("checkpointLocation", "/Volumes/checkpoints1/").toTable("stdcatalog.stdschema.deltatable") 示例 2：不支持的代码 spark.readStream.option("withEventTimeOrder", "true").format("format") .table("stdcatalog.stdschema.samplecsv")
Kafka	支持任何 Kafka 兼容流，不含三部分命名惯例 根据三部分命名约定，不支持基于 Kafka 的目录 )
OCI Streaming 服务	受支持
OCI 对象存储路径（使用 OCI：//）	不支持
Oracle Autonomous AI Lakehouse 、Oracle AI Database 、Oracle Autonomous AI Transaction Processing	不支持流处理（readStream 或 writeStream）

使用记事本的结构化流处理

您可以编写 Python 代码来处理笔记本中的流数据。卷路径或工作区路径作为检查点位置有效，但对象存储路径（oci：// 格式）不支持作为检查点位置。我们建议将卷路径用作检查点位置。

在运行流代码时，您可以从记事本中的仪表盘选项卡中查看与 Apache Spark 流相关的事件，例如输入速率、处理速率和批处理持续时间。

您还可以在逐步开发代码时从原始数据选项卡中查看与原始流相关的事件。

使用工作流配置 Spark 结构化流处理

您可以在工作流内配置流任务，以便连续处理流数据。

您首先需要创建一个作业，然后向该作业添加一个记事本或 Python 任务，以开始在 Oracle AI Data Platform Workbench 中使用流处理工作流。

1. 导航到工作区，然后单击工作流。
2. 单击 Create Job（创建作业）。
3. 为作业提供名称和说明。
4. 单击浏览，然后选择要将作业保存在 AI 数据平台工作台中的位置。单击选择。
5. 为最大并发运行数输入 1 。
6. 单击创建。
7. 单击刚刚创建的作业。
8. 单击添加任务。
9. 为您的任务提供名称。
10. 为任务类型选择记事本或 Python 。
11. 单击浏览并导航到要作为流处理任务添加的记事本或 Python 脚本。单击选择。
12. 为 Notebook 或 Python 任务选择计算集群（如果尚未附加计算集群）。
13. 选中流处理复选框。选择“流处理”将禁用执行超时和任务相关性作为选项。
    
    

    

    
    
14. 选择任务在失败时应尝试的重试次数。如果选择 0 以上，还必须指定作业在两次重试之间应等待多长时间，以及是否应在超时时尝试重试。
    
    

    

    
    
15. 单击立即运行。

流处理任务启动后，它会继续运行，直到您手动停止它。在每月定期维护期间，服务会停止并重新启动流处理任务，而无需您执行任何操作。