Dagster是一個強大的數據編排平臺，它提供了多種工具來幫助數據工程師構建可靠的數據管道。在Dagster中，Ops和Assets是兩種核心概念，用于定義數據處理邏輯。本文將全面介紹Ops的概念、特性及其使用方法，特別補充了Op上下文和Op工廠等重要內容，并解釋為什么對于新用戶我們推薦優先使用Assets。

什么是Ops？

Ops是Dagster中的基本計算單元，代表一個獨立的數據處理任務。每個Op應該執行相對簡單的任務，例如：

• 從其他數據集派生新數據集
• 執行數據庫查詢
• 在遠程集群中啟動Spark作業
• 查詢API并將結果存儲到數據倉庫
• 發送電子郵件或Slack消息

Ops的核心特性

1. 靈活的執行策略

Ops是獨立于執行策略的邏輯單元，這使得它們可以在開發和生產環境之間無縫轉換。Ops可以組合成圖(graphs)，并通過jobs綁定到適當的執行器上，實現單機執行或在集群中分布式執行。

2. 可插拔的外部系統集成

對于需要與外部系統交互的數據管道，Dagster提供了資源(resources)抽象層。你可以針對抽象資源(如數據庫)編寫Op邏輯，然后在job級別綁定具體的資源定義。這樣，開發階段可以使用本地替代方案，而生產環境則使用云服務。

3. 輸入和輸出管理

Ops具有明確的輸入和輸出，類似于Python函數的參數和返回值。這些輸入輸出可以附加Dagster類型進行運行時驗證，并可以通過IO Manager管理數據存儲，實現不同執行環境間的I/O策略切換和中間數據的高效緩存。

4. 配置能力

數據管道中的操作通常需要參數化配置。Ops允許通過配置模式(config schema)定義這些參數，使Ops更加靈活和可重用。例如，可以通過配置指定API端點：

from dagster import Configclass MyOpConfig(Config):api_endpoint: str@op
def my_configurable_op(config: MyOpConfig):data = requests.get(f"{config.api_endpoint}/data").json()return data

5. 事件流

Ops在執行過程中會發出一系列事件，包括默認事件(如開始執行)和通過事件API報告的自定義事件(如數據資產創建、數據質量檢查結果等)。這些事件流可以在Dagster UI中可視化，便于調試、檢查和實時監控。

6. 可測試性

Ops的設計使其易于測試，可以單獨測試或在管道中測試。資源API還允許在需要時替換外部系統(如數據庫)的存根(stub)。

定義和使用Ops

使用@op裝飾器定義Ops：

@op
def my_op():return "hello"

輸入和輸出

Ops通過參數接收輸入，通過返回值產生輸出：

@op
def add(a: int, b: int) -> int:return a + b

對于多輸出，可以使用Out對象：

@op(out={"sum": Out(), "product": Out()})
def math_ops(a: int, b: int):yield Output(a + b, "sum")yield Output(a * b, "product")

配置

為Ops添加配置：

from dagster import Config, opclass GreetingConfig(Config):name: str@op(config_schema=GreetingConfig)
def greet(context, config: GreetingConfig):context.log.info(f"Hello, {config.name}!")return f"Hello, {config.name}!"

Op上下文（Op Context）

在編寫Op時，用戶可以可選地提供一個上下文參數（通常命名為context）。當這個參數被提供時，Dagster會在Op執行時自動注入一個上下文對象，該對象提供了訪問系統信息的能力，如日志記錄器、當前運行ID等。

上下文對象的作用

1. 日志記錄：通過context.log記錄不同級別的日志信息
2. 訪問運行信息：獲取當前運行的ID、作業名稱等信息
3. 資源訪問：在某些情況下可以訪問配置的資源
4. 錯誤處理：提供更豐富的錯誤報告能力

使用示例

from dagster import op, OpExecutionContext@op
def context_op(context: OpExecutionContext):# 記錄info級別日志context.log.info(f"My run ID is {context.run_id}")# 記錄debug級別日志（默認可能不顯示）context.log.debug("This is a debug message")# 在實際業務邏輯中使用上下文try:result = do_something()return resultexcept Exception as e:context.log.error(f"Operation failed: {str(e)}")raise

Op工廠模式

在實際項目中，我們經常需要創建多個相似的Ops，或者需要動態生成Ops。這時，Op工廠模式就非常有用。Op工廠允許我們通過函數來生成Ops，而不是為每個Op手動編寫裝飾器。

工廠模式的應用場景

1. 參數化Op創建：當需要創建多個相似但配置不同的Ops時
2. 動態Op生成：根據運行時條件或配置動態生成Ops
3. 代碼復用：避免重復的Op定義代碼

創建Op工廠

from dagster import op, OpDefinitiondef create_math_op(name: str, operation):"""創建數學運算Op的工廠函數Args:name (str): 新Op的名稱operation (callable): 數學運算函數Returns:OpDefinition: 生成的Op定義"""@op(name=name)def math_op(a: float, b: float) -> float:return operation(a, b)return math_op# 使用工廠創建具體的Ops
add_op = create_math_op("add", lambda a, b: a + b)
multiply_op = create_math_op("multiply", lambda a, b: a * b)# 或者更復雜的工廠函數
def advanced_op_factory(config_schema=None, tags=None):"""更高級的Op工廠，支持配置和標簽Args:config_schema: Op的配置模式tags: 要附加到Op的標簽Returns:函數：接受compute函數并返回OpDefinition"""def decorator(compute_fn):op_def = op(name=compute_fn.__name__,config_schema=config_schema,tags=tags)(compute_fn)return op_defreturn decorator# 使用高級工廠
@advanced_op_factory(config_schema={"precision": int},tags={"team": "analytics"}
)
def divide(a: float, b: float, context) -> float:precision = context.op_config["precision"]result = a / breturn round(result, precision)

工廠模式的注意事項

1. 性能考慮：工廠模式會引入額外的函數調用層，但在大多數情況下影響可以忽略
2. 類型提示：使用工廠創建的Ops可能需要額外的類型提示處理
3. 文檔：確保為工廠函數和生成的Ops提供清晰的文檔

為什么推薦Assets而非Ops？

雖然Ops功能強大，但對于新用戶我們推薦優先使用Assets：

1. 更高級的抽象：Assets提供了數據版本控制、血緣追蹤和自動緩存等高級功能
2. 聲明式API：Assets允許以更聲明式的方式定義數據管道
3. 內置集成：Assets與Dagster的其他功能(如調度、監控)有更好的集成
4. 簡化復雜性：對于大多數用例，Assets可以簡化數據管道的定義和維護

結論

Ops是Dagster中強大的計算單元，適合處理復雜的數據處理邏輯。然而，對于新用戶或構建標準數據管道的場景，Assets提供了更高級的抽象和更簡化的開發體驗。隨著對Dagster的深入理解，用戶可以根據需要選擇使用Ops來處理更復雜的場景。

無論選擇哪種方式，Dagster都提供了豐富的工具和靈活性來構建可靠、可維護的數據管道。特別是Op上下文和Op工廠模式等高級特性，為復雜的數據工程需求提供了強大的支持。