モダンデータスタックなワークフローオーケストレーションツール「Prefect」 試してみた

from prefect import flow, task

@task
def task1():
    print("task1")

@flow
def first_flow():
    task1()
    print("hello prefect")

if __name__ == "__main__":
    first_flow()

from airflow.utils.dates import days_ago
from airflow.decorators import task, dag

@task
def task_1():
    return 'first task'

default_args = {
    'owner': 'airflow',
    'start_date': days_ago(2),
}


@dag(dag_id='taskflow', default_args=default_args)
def my_dag():
    op_1 = task_1()

dag = my_dag()

import random
from prefect import flow, task

@task(name="task1")
def task1():
    print("task1")
    return random.randint(1,10)

@task
def task2():
    print('task2')

@task
def task3():
    print('task3')

@flow
def first_flow():
    num = task1()
    print(f"number:{num}")
    if num > 5:
        task2()
    else:
        task3()

    print("hello prefect")

if __name__ == "__main__":
    first_flow()

　　@task
    def random_fun():
        import random

        return random.randrange(-10, 10) > 0

    @task.branch()
    def branching(x):
        if x is True:
            return "true_branch"
        else:
            return "false_branch"

    @task()
    def true_branch():
        print("True")

    @task()
    def false_branch():
        print("False")

    branching(random_fun()) >> [true_branch(), false_branch()]

@task
def task_a():
    pass

@task
def task_b():
    pass

@task
def task_c():
    pass
    
@task
def task_d():
    pass

@flow
def my_flow():
    a = task_a.submit()
    b = task_b.submit()
    # Wait for task_a and task_b to complete
    c = task_c.submit(wait_for=[a, b])
    # task_d will wait for task_c to complete
    # Note: If waiting for one task it must still be in a list.
    d = task_d(wait_for=[c])

from airflow.utils.dates import days_ago
from airflow.decorators import task, dag


@task
def task_1():
    return 'first task'

@task
def task_2():
    return 'second task'

@task
def task_3(value1, value2):
    return 'third task'

@task
def task_4(value):
    return 'fourth task'

default_args = {
    'owner': 'airflow',
    'start_date': days_ago(2),
}


@dag(dag_id='taskflow', default_args=default_args)
def my_dag():
    op_1 = task_1()
    op_2 = task_2()
    op_3 = task_3(op_1, op_2)
    op_4 = task_4(op_3)

dag = my_dag()

@task(retries=3)
def some_task_with_manual_backoff_retries():

@task(retries=3)
def test_retries():

from prefect import flow, task

@task(name="task1")
def task1(x):
    print(x)
    return x + 2

@task(name="task2")
def task2(x):
    print(x)
    return 2*x

@task(name="task3")
def task3(x):
    print(x)
    return x**2

@flow(name="first flow")
def first_flow(num):
    num1 = task1(num)
    num2 = task2(num1)
    num3 = task3(num2)
    print(num3)

if __name__ == "__main__":
    first_flow(3)

from __future__ import annotations
import json
import pendulum
from airflow.decorators import dag, task

@dag(
    schedule=None,
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
    catchup=False,
    tags=["example"],
)
def tutorial_taskflow_api():

    @task()
    def extract():
        data_string = '{"1001": 301.27, "1002": 433.21, "1003": 502.22}'

        order_data_dict = json.loads(data_string)
        return order_data_dict

    @task(multiple_outputs=True)
    def transform(order_data_dict: dict):
        total_order_value = 0

        for value in order_data_dict.values():
            total_order_value += value

        return {"total_order_value": total_order_value}

    @task()
    def load(total_order_value: float):
        print(f"Total order value is: {total_order_value:.2f}")

    order_data = extract()
    order_summary = transform(order_data)
    load(order_summary["total_order_value"])


tutorial_taskflow_api()