Durable Functions での関数チェーン - Hello シーケンスサンプル

[アーティクル]
03/28/2023

関数チェーンとは、特定の順序で一連の関数を実行するパターンです。ある関数の出力が、別の関数の入力に適用される必要がある、ということがよくあります。この記事では、Durable Functions のクイックスタート (C#、JavaScript、TypeScript、Python、PowerShell、Java) を実行するときに作成するチェーンのシーケンスについて説明します。 Durable Functions について詳しくは、「Durable Functions overview」(Durable Functions の概要) をご覧ください。

前提条件

Note

Azure Functions の Node.js プログラミングモデルのバージョン 4 は一般提供されています。新しい v4 モデルは、JavaScript と TypeScript の開発者にとって、より柔軟で直感的なエクスペリエンスが得られるように設計されています。 v3 と v4 の違いの詳細については、移行ガイドを参照してください。

次のコードスニペットでは、JavaScript (PM4) は、新しいエクスペリエンスであるプログラミングモデル V4 を示しています。

関数

この記事では、サンプルアプリで使用されている次の関数について説明します。

E1_HelloSequence:1 つのオーケストレーター機能。1 つのシーケンスで E1_SayHello を複数回呼び出します。 E1_SayHello 呼び出しからの出力を格納し、結果を記録します。
E1_SayHello:文字列の先頭に "Hello" を付加するアクティビティ関数。
HttpStart:オーケストレーターのインスタンスを起動する HTTP によってトリガーされる永続的なクライアント関数。

E1_HelloSequence オーケストレーター関数

[FunctionName("E1_HelloSequence")]
public static async Task<List<string>> Run(
    [OrchestrationTrigger] IDurableOrchestrationContext context)
{
    var outputs = new List<string>();

    outputs.Add(await context.CallActivityAsync<string>("E1_SayHello", "Tokyo"));
    outputs.Add(await context.CallActivityAsync<string>("E1_SayHello", "Seattle"));
    outputs.Add(await context.CallActivityAsync<string>("E1_SayHello_DirectInput", "London"));

    // returns ["Hello Tokyo!", "Hello Seattle!", "Hello London!"]
    return outputs;
}

すべての C# オーケストレーション関数は、Microsoft.Azure.WebJobs.Extensions.DurableTask アセンブリにある DurableOrchestrationContext 型のパラメーターを持つ必要があります。このコンテキストオブジェクトでは、他のアクティビティ関数を呼び出し、その CallActivityAsync メソッドを使用して入力パラメーターを渡すことができます。

このコードでは、E1_SayHello を異なるパラメーター値で 3 回続けて呼び出します。各呼び出しの戻り値が outputs 一覧に追加され、それが関数の末尾に返されます。

function.json

Visual Studio Code または Azure Portal を開発に使用する場合は、こちらのオーケストレーター関数の function.json ファイルの内容をご覧ください。ほとんどの orchestrator function.json ファイルは、このような内容です。

{
  "bindings": [
    {
      "name": "context",
      "type": "orchestrationTrigger",
      "direction": "in"
    }
  ],
  "disabled": false
}

重要な点は、orchestrationTrigger というバインドの種類です。すべての orchestrator 機能は、このトリガーの種類を使用する必要があります。

警告

Orchestrator 機能の "I/O なし" の規則に従うには、orchestrationTrigger トリガーバインドを使用する場合に、入力または出力バインドを使用しないでください。他の入力または出力バインドが必要な場合は、activityTrigger 関数のコンテキストで使用する必要があります。それらがオーケストレーターによって呼び出されます。詳細については、「オーケストレーター関数コードの制約」の記事を参照してください。

index.js

オーケストレーター関数をこちらに示します。

const df = require("durable-functions");

module.exports = df.orchestrator(function* (context) {
    context.log("Starting chain sample");
    const output = [];
    output.push(yield context.df.callActivity("E1_SayHello", "Tokyo"));
    output.push(yield context.df.callActivity("E1_SayHello", "Seattle"));
    output.push(yield context.df.callActivity("E1_SayHello", "London"));

    return output;
});

すべての JavaScript オーケストレーション関数に、durable-functions module モジュールが含まれている必要があります。それは、Durable Functions を JavaScript で記述することができるライブラリです。オーケストレーター関数と他の JavaScript 関数には、次の 3 つの大きな違いがあります。

オーケストレーター関数はジェネレーター関数です。
この関数は、durable-functions モジュールの orchestrator メソッドの呼び出しにラップされます (ここでは df)。
この関数は同期的であることが必要です。 "orchestrator" メソッドによって "context.done" の最終呼び出しが処理されるため、この関数は "値を返す" ことのみを行います。

context オブジェクトには、df 永続的なオーケストレーションコンテキストオブジェクトが含まれています。このオブジェクトを使用すると、他の アクティビティ 関数を呼び出し、その callActivity メソッドを使用して入力パラメーターを渡すことができます。このコードでは、異なるパラメーター値で E1_SayHello を 3 回続けて呼び出しています。yield を使用して実行を示すと、非同期アクティビティ関数呼び出しが返されるのを待つ必要があります。各呼び出しの戻り値が outputs 配列に追加されます。これは、関数の末尾に返されます。

const df = require("durable-functions");

const helloActivityName = "sayHello";

df.app.orchestration("helloSequence", function* (context) {
    context.log("Starting chain sample");

    const output = [];
    output.push(yield context.df.callActivity(helloActivityName, "Tokyo"));
    output.push(yield context.df.callActivity(helloActivityName, "Seattle"));
    output.push(yield context.df.callActivity(helloActivityName, "Cairo"));

    return output;
});

すべての JavaScript オーケストレーション関数に、durable-functions module モジュールが含まれている必要があります。このモジュールにより、Durable Functions を JavaScript で記述することができます。 V4 ノードプログラミングモデルを使用するには、プレビュー v3.x バージョンの durable-functions をインストールする必要があります。

オーケストレーター関数と他の JavaScript 関数には、次の 2 つの大きな違いがあります。

オーケストレーター関数はジェネレーター関数です。
この関数は同期的であることが必要です。関数は単純に 'return' する必要があります。

context オブジェクトには、df 永続的なオーケストレーションコンテキストオブジェクトが含まれています。このオブジェクトを使用すると、他の アクティビティ 関数を呼び出し、その callActivity メソッドを使用して入力パラメーターを渡すことができます。このコードでは、異なるパラメーター値で sayHello を 3 回続けて呼び出しています。yield を使用して実行を示すと、非同期アクティビティ関数呼び出しが返されるのを待つ必要があります。各呼び出しの戻り値が outputs 配列に追加されます。これは、関数の末尾に返されます。

Note

Python Durable Functions は、Functions 3.0 ランタイムでのみ利用できます。

function.json

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "context",
      "type": "orchestrationTrigger",
      "direction": "in"
    }
  ]
}

重要な点は、orchestrationTrigger というバインドの種類です。すべての orchestrator 機能は、このトリガーの種類を使用する必要があります。

警告

init.py

オーケストレーター関数をこちらに示します。

import azure.functions as func
import azure.durable_functions as df


def orchestrator_function(context: df.DurableOrchestrationContext):
    result1 = yield context.call_activity('E1_SayHello', "Tokyo")
    result2 = yield context.call_activity('E1_SayHello', "Seattle")
    result3 = yield context.call_activity('E1_SayHello', "London")
    return [result1, result2, result3]

main = df.Orchestrator.create(orchestrator_function)

すべての Python オーケストレーション関数に、durable-functions パッケージが含まれている必要があります。これは、Durable Functions を Python で記述できるようにするライブラリです。オーケストレーター関数と他の Python 関数には、次の 2 つの大きな違いがあります。

オーケストレーター関数はジェネレーター関数です。
この "ファイル" では、ファイルの末尾に main = df.Orchestrator.create(<orchestrator function name>) を指定することによって、オーケストレーター関数をオーケストレーターとして登録しています。これにより、ファイルで宣言されている他のヘルパー関数と区別できます。

context オブジェクトでは、他の "アクティビティ" 関数を呼び出し、その call_activity メソッドを使用して入力パラメーターを渡すことができます。このコードでは、異なるパラメーター値で E1_SayHello を 3 回続けて呼び出しています。yield を使用して実行を示すと、非同期アクティビティ関数呼び出しが返されるのを待つ必要があります。各呼び出しの戻り値は関数の末尾に返されます。

E1_SayHello アクティビティ関数

[FunctionName("E1_SayHello")]
public static string SayHello([ActivityTrigger] IDurableActivityContext context)
{
    string name = context.GetInput<string>();
    return $"Hello {name}!";
}

アクティビティは、ActivityTrigger 属性を使用します。指定された IDurableActivityContext を使用して、GetInput<T> を使用した入力値へのアクセスなど、アクティビティ関連のアクションを実行します。

E1_SayHello の実装は、比較的単純な文字列の書式設定操作です。

IDurableActivityContextにバインドする代わりに、アクティビティ関数に渡される型に直接バインドすることができます。次に例を示します。

[FunctionName("E1_SayHello_DirectInput")]
public static string SayHelloDirectInput([ActivityTrigger] string name)
{
    return $"Hello {name}!";
}

E1_SayHello/function.json

アクティビティ関数 E1_SayHello の function.json ファイルは、E1_HelloSequence のそれに似ていますが、バインドの種類 orchestrationTrigger の代わりにバインドの種類 activityTrigger を使用する点が違います。

{
  "bindings": [
    {
      "name": "name",
      "type": "activityTrigger",
      "direction": "in"
    }
  ],
  "disabled": false
}

Note

オーケストレーション関数によって呼び出されるすべてのアクティビティ関数は、activityTrigger バインドを使用する必要があります。

E1_SayHello の実装は、比較的単純な文字列の書式設定操作です。

E1_SayHello/index.js

module.exports = function (context) {
    context.done(null, `Hello ${context.bindings.name}!`);
};

オーケストレーション関数とは異なり、アクティビティ関数には特別な設定は不要です。オーケストレーター関数によって渡される入力は、activityTrigger バインドという名前で context.bindings オブジェクト (この例では context.bindings.name) に配置されます。サンプルコードに示すように、バインド名はエクスポートされた関数のパラメーターとして設定し、直接アクセスできます。

sayHello の実装は、比較的単純な文字列の書式設定操作です。

const df = require("durable-functions");

const helloActivityName = "sayHello";

df.app.activity(helloActivityName, {
    handler: function (input) {
        return `Hello ${input}`;
    },
});

オーケストレーション関数とは異なり、アクティビティ関数には特別な設定は不要です。オーケストレーター関数によって渡される入力は、関数の最初の引数です。 2 番目の引数は呼び出しコンテキストです。この例では使用されていません。

E1_SayHello/function.json

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "name",
      "type": "activityTrigger",
      "direction": "in"
    }
  ]
}

Note

オーケストレーション関数によって呼び出されるすべてのアクティビティ関数は、activityTrigger バインドを使用する必要があります。

E1_SayHello の実装は、比較的単純な文字列の書式設定操作です。

E1_SayHello/init.py

def main(name: str) -> str:
    return f"Hello {name}!"

オーケストレーター関数とは異なり、アクティビティ関数には特別な設定は不要です。オーケストレーター関数によって渡された入力には、関数のパラメーターとして直接アクセスできます。

HttpStart クライアント関数

オーケストレーター関数のインスタンスを開始するには、クライアント関数を使用します。 E1_HelloSequence のインスタンスを開始するには、HttpStart HTTP によってトリガーされる関数を使用します。

public static class HttpStart
{
    [FunctionName("HttpStart")]
    public static async Task<HttpResponseMessage> Run(
        [HttpTrigger(AuthorizationLevel.Function, methods: "post", Route = "orchestrators/{functionName}")] HttpRequestMessage req,
        [DurableClient] IDurableClient starter,
        string functionName,
        ILogger log)
    {
        // Function input comes from the request content.
        object eventData = await req.Content.ReadAsAsync<object>();
        string instanceId = await starter.StartNewAsync(functionName, eventData);

        log.LogInformation($"Started orchestration with ID = '{instanceId}'.");

        return starter.CreateCheckStatusResponse(req, instanceId);
    }
}

オーケストレーター操作をするには、関数に DurableClient 入力バインドが含まれている必要があります。クライアントを使用して、オーケストレーションを開始します。また、新しいオーケストレーションの状態を確認するための URL を含む HTTP 応答を返すこともできます。

HttpStart/function.json

{
    "bindings": [
      {
        "authLevel": "anonymous",
        "name": "req",
        "type": "httpTrigger",
        "direction": "in",
        "route": "orchestrators/{functionName}",
        "methods": ["post"]
      },
      {
        "name": "$return",
        "type": "http",
        "direction": "out"
      },
      {
        "name": "starter",
        "type": "orchestrationClient",
        "direction": "in"
      }
    ],
    "disabled": false
  }

オーケストレーター操作をするには、関数に durableClient 入力バインドが含まれている必要があります。

HttpStart/index.js

const df = require("durable-functions");

module.exports = async function (context, req) {
    const client = df.getClient(context);
    const instanceId = await client.startNew(req.params.functionName, undefined, req.body);

    context.log(`Started orchestration with ID = '${instanceId}'.`);

    return client.createCheckStatusResponse(context.bindingData.req, instanceId);
};

DurableOrchestrationClient オブジェクトを取得するには、df.getClient を使用します。クライアントを使用して、オーケストレーションを開始します。また、新しいオーケストレーションの状態を確認するための URL を含む HTTP 応答を返すこともできます。

const df = require("durable-functions");
const { app } = require("@azure/functions");

app.http("httpStart", {
    route: "orchestrators/{orchestratorName}",
    extraInputs: [df.input.durableClient()],
    handler: async (request, context) => {
        const client = df.getClient(context);
        const body = await request.json();
        const instanceId = await client.startNew(request.params.orchestratorName, { input: body });

        context.log(`Started orchestration with ID = '${instanceId}'.`);

        return client.createCheckStatusResponse(request, instanceId);
    },
});

オーケストレーターを管理および操作するには、関数に durableClient 入力バインディングが必要です。このバインディングは、関数を登録するときに extraInputs 引数で指定する必要があります。 durableClient 入力は、df.input.durableClient() を呼び出すことによって取得できます。

DurableClient オブジェクトを取得するには、df.getClient を使用します。クライアントを使用して、オーケストレーションを開始します。また、新しいオーケストレーションの状態を確認するための URL を含む HTTP 応答を返すこともできます。

HttpStart/function.json

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "authLevel": "anonymous",
      "name": "req",
      "type": "httpTrigger",
      "direction": "in",
      "route": "orchestrators/{functionName}",
      "methods": [
        "post",
        "get"
      ]
    },
    {
      "name": "$return",
      "type": "http",
      "direction": "out"
    },
    {
      "name": "starter",
      "type": "durableClient",
      "direction": "in"
    }
  ]
}

オーケストレーター操作をするには、関数に durableClient 入力バインドが含まれている必要があります。

HttpStart/init.py

import logging

import azure.functions as func
import azure.durable_functions as df


async def main(req: func.HttpRequest, starter: str) -> func.HttpResponse:
    client = df.DurableOrchestrationClient(starter)
    instance_id = await client.start_new(req.route_params["functionName"], None, None)

    logging.info(f"Started orchestration with ID = '{instance_id}'.")

    return client.create_check_status_response(req, instance_id)

DurableOrchestrationClient を使用して、Durable Functions クライアントを取得します。クライアントを使用して、オーケストレーションを開始します。また、新しいオーケストレーションの状態を確認するための URL を含む HTTP 応答を返すこともできます。

サンプルを実行する

E1_HelloSequence オーケストレーションを実行するには、次の HTTP POST 要求を HttpStart 関数に送信します。

POST http://{host}/orchestrators/E1_HelloSequence

Note

前の HTTP スニペットでは、既定の api/ プレフィックスをすべての HTTP トリガー関数 URL から削除するエントリが host.json ファイルにあることを想定しています。この構成のマークアップはサンプルの host.json ファイルにあります。

たとえば、"myfunctionapp" という関数アプリのサンプルを実行する場合、"{host}" を "myfunctionapp.azurewebsites.net" に置き換えます。

結果は次のような HTTP 202 応答です (簡潔にするため省略しています)。

HTTP/1.1 202 Accepted
Content-Length: 719
Content-Type: application/json; charset=utf-8
Location: http://{host}/runtime/webhooks/durabletask/instances/96924899c16d43b08a536de376ac786b?taskHub=DurableFunctionsHub&connection=Storage&code={systemKey}

(...trimmed...)

この時点で、オーケストレーションはキューに登録され、すぐに実行を開始します。 Location ヘッダー内の URL は、実行状態を確認するのに使用できます。

GET http://{host}/runtime/webhooks/durabletask/instances/96924899c16d43b08a536de376ac786b?taskHub=DurableFunctionsHub&connection=Storage&code={systemKey}

この結果がオーケストレーションの状態です。迅速に実行して完了し、次のような応答で完了状態にあることがわかります(簡潔にするため省略しています)。

HTTP/1.1 200 OK
Content-Length: 179
Content-Type: application/json; charset=utf-8

{"runtimeStatus":"Completed","input":null,"output":["Hello Tokyo!","Hello Seattle!","Hello London!"],"createdTime":"2017-06-29T05:24:57Z","lastUpdatedTime":"2017-06-29T05:24:59Z"}

このように、インスタンスの runtimeStatus は完了しており、output は orchestrator 関数の実行の JSON でシリアル化された結果を格納します。

Note

同様のスターターロジックを queueTrigger、eventHubTrigger、timerTrigger などの他のトリガーの種類に実装することができます。

関数の実行ログを確認しましょう。 E1_HelloSequence 関数が、オーケストレーションの信頼性に関するトピックで説明されている再生の動作のため、複数回、開始し完了しています。その一方で、E1_SayHello の実行は 3 回だけでした。これらの関数の実行は再生されなかったためです。

次のステップ

このサンプルでは、単純な関数チェーンのオーケストレーションについて説明しました。次のサンプルでは、ファンアウト/ファンインパターンの実装方法について説明します。

ファンアウト/ファンインサンプルの実行

Durable Functions での関数チェーン - Hello シーケンス サンプル

前提条件

関数

E1_HelloSequence オーケストレーター関数

E1_SayHello アクティビティ関数

HttpStart クライアント関数

サンプルを実行する

次のステップ

その他のリソース

Durable Functions での関数チェーン - Hello シーケンスサンプル