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

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

Python v1 プログラミング モデルでは、関数フォルダー内の個別の function.json ファイルでバインドを定義する必要があります。 詳細については、「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;
}

次の例は、メソッドの戻り値を出力として使用してメッセージをイベント ハブに書き込む 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");
    }
}

タイマーでトリガーされ、イベント ハブに 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}`];

TypeScript のサンプルは、モデル 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();
};

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")

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

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

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

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

インプロセス 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 の出力バインドは次の型にバインドできます。

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

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

カスタム型を定義するか、文字列を使用する必要があります。 拡張機能 v5.x 以降では、追加のオプションを使用できます。

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

context.bindings.<name> を使用して出力イベントにアクセスします。<name> は context.bindings.<name> の name プロパティで指定された値です。