11 串流處理

本節涵蓋在 Oracle AI Data Platform Workbench 中使用串流資料或持續產生的資料。

主題：

• 關於串流處理
• 使用工作流程設定 Spark 結構化串流處理

關於串流處理

您可以在 Oracle AI Data Platform Workbench 中使用 Apache Spark Structured Streaming 功能，以近乎即時的方式處理串流資料或持續產生資料。

筆記型電腦和工作流程都支援 Apache Spark 結構化串流處理。您可以使用下列來源和接收器來讀取串流資料、將串流資料寫入檢查點位置和檢查點位置。

表格 11-1 支援的來源與接收器

來源或槽 Stencils	支援服務？
磁碟區路徑 (/Volume/bronze/bucket1)	支援所有格式
工作區路徑 (/Workspace/folder1/)	支援所有格式
目錄中含有三個零件名稱的表格 (catalog.schema.table)	僅支援差異格式 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. 按一下 建立工作。
3. 提供您工作的名稱和描述。
4. 按一下瀏覽，然後選取要將工作儲存在「AI 資料平台工作台」中的位置。按一下「選取」。
5. 輸入 1 代表並行執行上限。
6. 按一下建立。
7. 按一下您剛建立的工作。
8. 按一下新增工作。
9. 請提供作業的名稱。
10. 選取記事本或 Python 作為工作類型。
11. 按一下瀏覽，然後瀏覽至要新增為「串流」作業的記事本或 Python 命令檔。按一下「選取」。
12. 選取記事本或 Python 作業的運算叢集 (如果尚未連附的話)。
13. 選取串流核取方塊。選取「串流處理」會停用執行逾時與作業相依性作為選項。
    
    

    

    
    
14. 選取工作失敗時應嘗試的重試次數。如果選取超過 0，您還必須指定工作執行在重試之間應該等待多少時間，以及是否應該在逾時時嘗試重試。
    
    

    

    
    
15. 按一下立即執行。

Streaming 工作啟動後，會繼續執行，直到您手動停止該工作為止。在每月定期維護期間，服務會停止並重新啟動 Streaming 作業，而不需要結束任何動作。