Azure Functions の Azure Event Hubs 出力バインディング

[アーティクル]
09/25/2023

この記事では、Azure Functions で Azure Event Hubs のバインドを使用する方法について説明します。 Azure Functions は、イベントハブのトリガーおよび出力バインドをサポートしています。

セットアップと構成の詳細については、概要に関するページをご覧ください。

Event Hubs 出力バインドを使用して、イベントストリームにイベントを書き込みます。イベントを書き込むには、イベントハブへの送信アクセス許可が必要です。

出力バインディングを実装する前に、必要なパッケージ参照が用意されていることを確認してください。

重要

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

Azure Functions では、Python の 2 つのプログラミングモデルがサポートされています。バインドを定義する方法は、選択したプログラミングモデルによって異なります。

Python v2 プログラミングモデルでは、Python 関数コードでデコレーターを使用してバインドを直接定義できます。詳細については、「Python 開発者ガイド」を参照してください。

この記事は、両方のプログラミングモデルをサポートしています。

次の例は、メソッドの戻り値を出力として使用してメッセージ文字列をイベントハブに書き込む C# 関数を示しています。

[Function(nameof(EventHubFunction))]
[FixedDelayRetry(5, "00:00:10")]
[EventHubOutput("dest", Connection = "EventHubConnection")]
public string EventHubFunction(
    [EventHubTrigger("src", Connection = "EventHubConnection")] string[] input,
    FunctionContext context)
{
    _logger.LogInformation("First Event Hubs triggered message: {msg}", input[0]);

    var message = $"Output message created at {DateTime.Now}";
    return message;
}

重要

インプロセスモデルのサポートは 2026 年 11 月 10 日に終了します。完全なサポートのために、アプリを分離ワーカーモデルに移行することを強くお勧めします。

次の例は、メソッドの戻り値を出力として使用してメッセージをイベントハブに書き込む C# 関数を示しています。

[FunctionName("EventHubOutput")]
[return: EventHub("outputEventHubMessage", Connection = "EventHubConnectionAppSetting")]
public static string Run([TimerTrigger("0 */5 * * * *")] TimerInfo myTimer, ILogger log)
{
    log.LogInformation($"C# Timer trigger function executed at: {DateTime.Now}");
    return $"{DateTime.Now}";
}

次の例では、IAsyncCollector インターフェイスを使用してメッセージのバッチを送信する方法を示します。これは、1 つのイベントハブから送信されるメッセージを処理し、その結果を別のイベントハブに送信する場合の一般的なシナリオです。

[FunctionName("EH2EH")]
public static async Task Run(
    [EventHubTrigger("source", Connection = "EventHubConnectionAppSetting")] EventData[] events,
    [EventHub("dest", Connection = "EventHubConnectionAppSetting")]IAsyncCollector<EventData> outputEvents,
    ILogger log)
{
    foreach (EventData eventData in events)
    {
        // Do some processing:
        string newEventBody = DoSomething(eventData);

        // Queue the message to be sent in the background by adding it to the collector.
        // If only the event is passed, an Event Hubs partition to be assigned via
        // round-robin for each batch.
        await outputEvents.AddAsync(new EventData(newEventBody));

        // If your scenario requires that certain events are grouped together in an
        // Event Hubs partition, you can specify a partition key.  Events added with 
        // the same key will always be assigned to the same partition.        
        await outputEvents.AddAsync(new EventData(newEventBody), "sample-key");
    }
}

モデル v4
モデル v3

タイマーでトリガーされ、イベントハブに 1 つのメッセージを送信する TypeScript 関数を次の例に示します。

import { app, InvocationContext, output, Timer } from '@azure/functions';

export async function timerTrigger1(myTimer: Timer, context: InvocationContext): Promise<string> {
    const timeStamp = new Date().toISOString();
    return `Message created at: ${timeStamp}`;
}

app.timer('timerTrigger1', {
    schedule: '0 */5 * * * *',
    return: output.eventHub({
        eventHubName: 'myeventhub',
        connection: 'MyEventHubSendAppSetting',
    }),
    handler: timerTrigger1,
});

複数のメッセージを出力するには、1 つのオブジェクトではなく配列を返します。次に例を示します。

const timeStamp = new Date().toISOString();
const message = `Message created at: ${timeStamp}`;
return [`1: ${message}`, `2: ${message}`];

モデル v4
モデル v3

タイマーでトリガーされ、イベントハブに 1 つのメッセージを送信する JavaScript 関数を次の例に示します。

const { app, output } = require('@azure/functions');

const eventHubOutput = output.eventHub({
    eventHubName: 'myeventhub',
    connection: 'MyEventHubSendAppSetting',
});

app.timer('timerTrigger1', {
    schedule: '0 */5 * * * *',
    return: eventHubOutput,
    handler: (myTimer, context) => {
        const timeStamp = new Date().toISOString();
        return `Message created at: ${timeStamp}`;
    },
});

複数のメッセージを出力するには、1 つのオブジェクトではなく配列を返します。次に例を示します。

const timeStamp = new Date().toISOString();
const message = `Message created at: ${timeStamp}`;
return [`1: ${message}`, `2: ${message}`];

次の例は、function.json ファイルの Event Hub トリガーバインドと、そのバインドが使用される関数を示しています。この関数では、出力メッセージをイベントハブに書き込みます。

function.json ファイルの Event Hubs バインディングデータの例を次に示します。これは、Functions ランタイムのバージョン 1.x とそれより後のバージョンとでは異なります。

{
    "type": "eventHub",
    "name": "outputEventHubMessage",
    "eventHubName": "myeventhub",
    "connection": "MyEventHubSendAppSetting",
    "direction": "out"
}

1 つのメッセージを送信する JavaScript コードを次に示します。

module.exports = function (context, myTimer) {
    var timeStamp = new Date().toISOString();
    context.log('Message created at: ', timeStamp);   
    context.bindings.outputEventHubMessage = "Message created at: " + timeStamp;
    context.done();
};

複数のメッセージを送信する JavaScript コードを次に示します。

module.exports = function(context) {
    var timeStamp = new Date().toISOString();
    var message = 'Message created at: ' + timeStamp;

    context.bindings.outputEventHubMessage = [];

    context.bindings.outputEventHubMessage.push("1 " + message);
    context.bindings.outputEventHubMessage.push("2 " + message);
    context.done();
};

PowerShell の完全な例は保留中です。

次の例は、イベントハブトリガーバインドと、そのバインドが使用される Python 関数を示しています。この関数では、メッセージをイベントハブに書き込みます。この例は、v1 と v2 のどちらの Python プログラミングモデルを使用するかによって異なります。

import logging
import azure.functions as func

app = func.FunctionApp()

@app.function_name(name="eventhub_output")
@app.route(route="eventhub_output")
@app.event_hub_output(arg_name="event",
                      event_hub_name="<EVENT_HUB_NAME>",
                      connection="<CONNECTION_SETTING>")
def eventhub_output(req: func.HttpRequest, event: func.Out[str]):
    body = req.get_body()
    if body is not None:
        event.set(body.decode('utf-8'))
    else:    
        logging.info('req body is none')
    return 'ok'

複数のメッセージを送信する Python コードを次に示します。

import logging
import azure.functions as func
from typing import List

app = func.FunctionApp()

@app.function_name(name="eventhub_output")
@app.route(route="eventhub_output")
@app.event_hub_output(arg_name="event",
                      event_hub_name="<EVENT_HUB_NAME>",
                      connection="<CONNECTION_SETTING>")

def eventhub_output(req: func.HttpRequest, event: func.Out[List[str]]) -> func.HttpResponse:
    my_messages=["message1", "message2","message3"]
    event.set(my_messages)
    return func.HttpResponse(f"Messages sent")

次の例は、function.json ファイル内の Event Hubs バインディングデータを示しています。

{
    "type": "eventHub",
    "name": "$return",
    "eventHubName": "myeventhub",
    "connection": "MyEventHubSendAppSetting",
    "direction": "out"
}

1 つのメッセージを送信する Python コードを次に示します。

import datetime
import logging
import azure.functions as func


def main(timer: func.TimerRequest) -> str:
    timestamp = datetime.datetime.utcnow()
    logging.info('Message created at: %s', timestamp)
    return 'Message created at: {}'.format(timestamp)

複数のメッセージを送信する Python コードを次に示します。

import logging
from typing import List
import azure.functions as func


def main(req: func.HttpRequest, messages:func.Out[List[str]]) -> func.HttpResponse:
    logging.info('Python HTTP trigger function processed a request.')
    messages.set([{"message1"}, {"message2"}])
    return func.HttpResponse(f"Messages sent")

次の例は、現在の時刻を含むメッセージをイベントハブに書き込む Java 関数を示しています。

@FunctionName("sendTime")
@EventHubOutput(name = "event", eventHubName = "samples-workitems", connection = "AzureEventHubConnection")
public String sendTime(
   @TimerTrigger(name = "sendTimeTrigger", schedule = "0 */5 * * * *") String timerInfo)  {
     return LocalDateTime.now().toString();
 }

Java 関数ランタイムライブラリで、その値が Event Hubs に公開されるパラメーター上で @EventHubOutput 注釈を使用します。パラメーターの型は OutputBinding<T> にする必要があります。T は POJO または Java の任意のネイティブ型です。

属性

インプロセスと分離ワーカープロセスの C# ライブラリはどちらも、属性を使用してバインドを構成します。 C# スクリプトでは、C# スクリプトガイドで説明されているように、代わりに function.json 構成ファイルを使用します。

分離ワーカーモデル
インプロセスモデル

[EventHubOutputAttribute] を使って、次のプロパティをサポートするイベントハブへの出力バインドを定義します。

パラメーター	説明
EventHubName	イベントハブの名前。イベントハブの名前は接続文字列にも存在し、その値が実行時にこのプロパティをオーバーライドします。
接続	Event Hubs への接続方法を指定するアプリ設定または設定コレクションの名前。詳細については、「接続」を参照してください

EventHubAttribute を使って、次のプロパティをサポートするイベントハブへの出力バインドを定義します。

パラメーター	説明
EventHubName	イベントハブの名前。イベントハブの名前は接続文字列にも存在し、その値が実行時にこのプロパティをオーバーライドします。
接続	Event Hubs への接続方法を指定するアプリ設定または設定コレクションの名前。詳細については、「接続」を参照してください

デコレータ

Python v2 プログラミングモデルにのみ適用されます。

デコレータを使用して定義された Python v2 関数の場合、cosmos_db_trigger に次のプロパティがあります。

プロパティ	説明
`arg_name`	イベントを表す関数コードに使用される変数の名前。
`event_hub_name`	イベントハブの名前。イベントハブの名前は接続文字列にも存在し、その値が実行時にこのプロパティをオーバーライドします。
`connection`	Event Hubs への接続方法を指定するアプリ設定または設定コレクションの名前。詳細については、「接続」を参照してください

function.json を使用して定義された Python 関数については、「構成」セクションを参照してください。

注釈

Java 関数ランタイムライブラリで、その値が Event Hubs に公開されるパラメーター上で EventHubOutput 注釈を使用します。注釈では、次の設定がサポートされています。

構成

"Python v1 プログラミングモデルにのみ適用されます。"

モデル v4
モデル v3

次の表では、output.eventHub() メソッドに渡される options オブジェクトに対して設定できるプロパティについて説明します。

プロパティ	説明
eventHubName	イベントハブの名前。イベントハブの名前は接続文字列にも存在し、その値が実行時にこのプロパティをオーバーライドします。
connection	Event Hubs への接続方法を指定するアプリ設定または設定コレクションの名前。詳細については、「接続」を参照してください

次の表は、function.json ファイルで設定したバインド構成のプロパティを説明しています。

プロパティ	説明
type	`eventHub` に設定する必要があります。
direction	`out` に設定する必要があります。このパラメーターは、Azure Portal でバインドを作成するときに自動で設定されます。
name	イベントを表す関数コードに使用される変数の名前。
eventHubName	イベントハブの名前。イベントハブの名前は接続文字列にも存在し、その値が実行時にこのプロパティをオーバーライドします。
connection	Event Hubs への接続方法を指定するアプリ設定または設定コレクションの名前。詳細については、「接続」を参照してください

次の表は、function.json ファイルで設定するバインド構成プロパティの説明です。ランタイムのバージョンごとに異なります。

Functions 2.x 以上
Functions 1.x

function.json のプロパティ	説明
type	`eventHub` に設定する必要があります。
direction	`out` に設定する必要があります。このパラメーターは、Azure Portal でバインドを作成するときに自動で設定されます。
name	イベントを表す関数コードに使用される変数の名前。
eventHubName	Functions 2.x 以降。イベントハブの名前。イベントハブの名前は接続文字列にも存在し、その値が実行時にこのプロパティをオーバーライドします。
connection	Event Hubs への接続方法を指定するアプリ設定または設定コレクションの名前。詳細については、「接続」を参照してください

function.json のプロパティ	説明
type	`eventHub` に設定する必要があります。
direction	`out` に設定する必要があります。このパラメーターは、Azure Portal でバインドを作成するときに自動で設定されます。
name	イベントを表す関数コードに使用される変数の名前。
path	Functions 1.x のみ。イベントハブの名前。イベントハブの名前は接続文字列にも存在し、その値が実行時にこのプロパティをオーバーライドします。
connection	Event Hubs への接続方法を指定するアプリ設定または設定コレクションの名前。詳細については、「接続」を参照してください

ローカルで開発する場合は、Values コレクション内の local.settings.json ファイルにアプリケーション設定を追加します。

使用法

Event Hubs 出力バインドでサポートされるパラメーター型は、Functions ランタイムのバージョン、拡張機能パッケージのバージョン、および使用される C# のモダリティによって異なります。

インプロセス C# クラスライブラリ関数では、次の型がサポートされています。

Azure.Messaging.EventHubs.EventData
String
Byte array
プレーンオールド CLR オブジェクト (POCO)

このバージョンの EventData では、EventBody を優先し、レガシ Body 型がサポートされなくなりました。

out string paramName などのメソッドパラメーターを使用してメッセージを送信します。複数のメッセージを書き込むには、out string の代わりに ICollector<EventData> または IAsyncCollector<EventData> を使用します。パーティションキーは、IAsyncCollector<EventData> でのみ使用できます。

インプロセス C# クラスライブラリ関数では、次の型がサポートされています。

Microsoft.Azure.EventHubs.EventData
String
Byte array
プレーンオールド CLR オブジェクト (POCO)

out string paramName などのメソッドパラメーターを使用してメッセージを送信します。複数のメッセージを書き込むには、out string の代わりに ICollector<string> または IAsyncCollector<string> を使用します。

関数で 1 つのイベントを書き込む場合、Event Hubs の出力バインドは次の型にバインドできます。

Type	説明
`string`	イベントを表す文字列。イベントが単純なテキストのときに使用します。
`byte[]`	イベントのバイト数。
JSON シリアル化可能な型	イベントを表すオブジェクト。 Functions は、単純な従来の CLR オブジェクト (POCO) 型を JSON データにシリアル化しようとします。

関数で複数のイベントを書き込むとき、Event Hubs 出力バインドは次の型にバインドできます。

Type	説明
`T[]` (`T` は単一のイベントの種類の 1 つ)	複数のイベントを含む配列。各エントリは 1 つのイベントを表します。

その他の出力シナリオでは、Microsoft.Azure.EventHubs から直接型を作成して使用します。

EventHubOutput 注釈を使用して関数から Event Hubs メッセージを出力するには、次の 2 つのオプションがあります。

戻り値: 関数自体に注釈を適用すると、関数の戻り値が Event Hubs メッセージとして永続化されます。
命令型: メッセージ値を明示的に設定するには、OutputBinding<T> 型の特定のパラメーターに注釈を適用します。ここで、T は POJO または任意のネイティブ Java 型です。この構成では、setValue メソッドに値を渡すと、その値が Event Hubs メッセージとして保持されます。

PowerShell の完全な例は保留中です。

モデル v4
モデル v3

値を直接返すか、context.extraOutputs.set() を使って、出力メッセージにアクセスします。

関数から Event Hubs メッセージを出力するには、次の 2 つのオプションがあります。

戻り値:function.json 内の name プロパティを $return に設定します。この構成では、関数の戻り値は Event Hubs メッセージとして永続化されます。
命令型:Out 型として宣言されたパラメーターの set メソッドに値を渡します。 set に渡された値は、Event Hubs メッセージとして永続化されます。

接続

connection プロパティは、アプリを Event Hubs に接続する方法を指定する環境構成への参照です。次が指定されている場合があります。

接続文字列を含むアプリケーション設定の名前
まとめて ID ベースの接続を定義する、複数のアプリケーション設定の共有プレフィックスの名前。

構成された値が、1 つの設定に完全一致し、プレフィックスがその他の設定とも一致する場合は、完全一致が使用されます。

接続文字列

この接続文字列を取得するには、イベントハブ自体ではなく、"名前空間" の [接続情報] をクリックします。接続文字列は、イベントハブ自体ではなく、Event Hubs 名前空間のものである必要があります。

トリガーに使用する場合、接続文字列には、機能を有効にするために少なくとも "読み取り" アクセス許可が必要です。出力バインディングに使用する場合、接続文字列には、イベントストリームにメッセージを送信するための "送信" アクセス許可が必要です。

この接続文字列は、バインディング構成の connection プロパティで指定した値と同じ名前のアプリケーション設定に格納する必要があります。

ID ベースの接続

シークレットを含む接続文字列を使う代わりに、拡張機能のバージョン 5.x 以降を使用している場合は、アプリで Microsoft Entra ID を使用できます。これを行うには、トリガーおよびバインド構成の connection プロパティにマップされる共通のプレフィックスに設定を定義します。

このモードでは、拡張機能に次のプロパティが必要です。

プロパティ	環境変数テンプレート	説明	値の例
完全修飾名前空間	`<CONNECTION_NAME_PREFIX>__fullyQualifiedNamespace`	完全修飾 Event Hubs の名前空間。	`myeventhubns.servicebus.windows.net`

接続をカスタマイズするには、プロパティを追加設定します。「ID ベース接続に共通のプロパティ」を参照してください。

注意

Azure App Configuration または Key Vault を使用してマネージド ID 接続の設定を指定する場合、__ の代わりに : や / などの有効なキー区切り記号を使用して、名前が正しく解決されるようにしなければなりません。

たとえば、「 <CONNECTION_NAME_PREFIX>:fullyQualifiedNamespace 」のように入力します。

Azure Functions サービスでホストされている場合、ID ベースの接続では、マネージド ID が使用されます。ユーザー割り当て ID を credential および clientID プロパティで指定できますが、システム割り当て ID が既定で使用されます。リソース ID を使用したユーザー割り当て ID の構成はサポートされていないことに注意してください。ローカル開発などの他のコンテキストで実行する場合は、代わりに開発者 ID が使用されますが、カスタマイズすることもできます。 ID ベースの接続によるローカル開発に関するページをご覧ください。

ID にアクセス許可を付与する

使用されている ID が何であれ、目的のアクションを実行するためのアクセス許可が必要です。ほとんどの Azure では、これはそれらのアクセス許可を提供する組み込みロールまたはカスタムロールを使って、Azure RBAC でロールを割り当てる必要があることを意味します。

重要

すべてのコンテキストに必要ではない一部のアクセス許可がターゲットサービスによって公開される場合があります。可能であれば、最小限の特権の原則に従い、必要な特権だけを ID に付与します。たとえば、アプリがデータソースからの読み取りのみを行う必要がある場合は、読み取りアクセス許可のみを持つロールを使用します。サービスへの書き込みも可能なロールを割り当てることは、読み取り操作に対するアクセス許可が過剰になるため、不適切です。同様に、ロールの割り当てが、読み取る必要のあるリソースだけに限定されていることを確認する必要があります。

実行時にイベントハブへのアクセスを提供するロールの割り当てを作成する必要があります。ロールの割り当てのスコープは、Event Hubs の名前空間に対するもの、またはイベントハブ自体になります。所有者のような管理ロールでは十分ではありません。次の表は、通常の操作で Event Hubs の拡張機能を使用するときに推奨される組み込みのロールを示しています。アプリケーションでは、記述したコードに基づいて追加のアクセス許可が必要になる場合があります。

[バインドの種類]	組み込みロールの例
トリガー	Azure Event Hubs データ受信者、Azure Event Hubs データ所有者
出力バインド	Azure Event Hubs データ送信者

例外とリターンコード

バインド	リファレンス
Event Hubs	運用ガイド

次のステップ

イベントハブのイベントストリームに送信されたイベントに応答する (トリガー)