終於到了倒數第二天了，那前一天我們介紹了用爬蟲作為資料獲取的手段，那今天就來稍微講解一下有了資料之後需要怎樣進行資料流(data pipeline)的管理吧

ETL概念

什麼是ETL？

ETL是一個縮寫，一共由以下幾個部分所組成

• Extract(抽取)
• Transform(轉換)
• Load(載入)

圖片來源：Microsoft

通常是用在資料的倉儲與管理，在越大量複雜的資料就越適合使用這種概念進行管理資料。

ETL有什麼幫助？

一筆原始的數據，通常會需要經過許多不同步驟的處理才會送入到我們最終需要的地方。而有的時候比起實作上是怎麼達成的細節，我們通常會更加在意這些經過不同步驟的階段的資料。因此將資料流在過程中以ETL的方式進行拆解和管理會可以更加有效的去維護和開發。

Dagster

Dagster是什麼？

• 一款可以定義資料元之間的依賴關係和流程的工具
• 可通過 API 查詢、操作和監控。
• 除了task執行的依賴性以外，另外針對data也建立依賴關係。
• 可以在支援多種其他工具或語言搭配使用
  • Spark
  • Python
  • SQL
  • Jupyter 
  • locally
  • CI/CD pipelines
  • staging
  • production
• 在python內用decorater建立流程
• 擁有很漂亮的Web UI介面用來操控和監視

Web介面，圖片來源：Dagster

安裝

安裝方式非常簡單，只要用pip像是安裝一般python套件一樣即可

pip install dagster dagit

• Dagster：核心元件，以及包含可以用命令行執行的工具
• Dagit：主要是Web UI的部分

基本使用

建立Dagster流程

我們建立一份Day29-1.py的檔案，內容如下

from dagster import execute_pipeline, pipeline, solid


@solid
def get_name():
    return 'dagster'


@solid
def hello(something):
    print(something)


@pipeline
def hello_pipeline():
    hello(get_name())

這邊我們一共用了兩種decrorater，其中

• solid：用來建立資料流的基本節點
• pipeline：在用來呼叫不同的Solid節點，Dagster會自動幫我們建立流程

開啟Dagit web介面

在Terminal輸入

dagit -f Day29-1.py

以單檔方式打開這份流程，然後開啟localhost:3000即可進入介面

Dagster幫忙建立好的流程圖

在Web介面上執行

有許多不同的執行流程顯示方式

進階一點點，執行時調整參數

dagster可以把一些參數留到Web UI的介面調整，因此我們這邊就來試看看，建立一份Day29-2.py

from dagster import pipeline, solid, InputDefinition, OutputDefinition, PresetDefinition, ModeDefinition


@solid(
    config_schema={"input": str}, # config輸入型態的檢查
    description="說明文字",

)
def string_from_config(context): 
    text = context.solid_config["input"] # 參數在.solid_config底下的dict
    return text


@solid(
    input_defs=[InputDefinition("input_string", dagster_type = str)], # 輸入變數的名稱和形態
) 
def show(context, input_string):
    context.log.info(input_string)
    return 1


@pipeline(
    mode_defs=[
        ModeDefinition(
            name="default1",
        ),
        ModeDefinition(
            name="default2",
        ),
    ],
    preset_defs=[
        # 預設config用
        PresetDefinition(
            "default1",
            run_config={
                "solids": {
                    "string_from_config": {"config": {"input": ":default string 1:"}}
                }
            },
            mode="default1",
        ),
        # 第二筆config
        PresetDefinition(
            "default2",
            run_config={
                "solids": {
                    "string_from_config": {"config": {"input": ":default string 2:"}}
                }
            },
            mode="default2",
        ),
    ],
)
def my_pipeline():
    show(string_from_config())

進入Playground會可以交出已經配置好的參數，或是自己另外更改

而結果也會隨著前面的配置而改動

結語

當初看到的時候是覺得這東西真的挺酷的，雖然上手門檻不低而且需要特地去適應他的邏輯，但習慣之後真的對一些資料流程專案的設計管理上是蠻有幫助的~