Триггер Apache Kafka для Функций Azure

Статья
01.06.2023

Триггер Apache Kafka можно использовать в Функциях Azure для запуска кода функции в ответ на сообщения в разделах Kafka. Вы также можете использовать выходную привязку Kafka для записи из функции в раздел. Сведения о настройке и конфигурации см. в статье Обзор привязок Apache Kafka для Функций Azure.

Важно!

Привязки Kafka доступны только для Функций в составе эластичного плана "Премиум" и плана "Выделенный (Служба приложений)". Они поддерживаются только в среде выполнения Функций версии 3.x и выше.

Пример

Использование триггера зависит от модальности C#, используемой в приложении-функции. Это может быть один из следующих вариантов:

Изолированная рабочая модель
Модель внутрипроцессного процесса

Изолированная библиотека классов рабочих процессов, скомпилированная функция C# выполняется в процессе, изолированном от среды выполнения.

Используемые атрибуты зависят от конкретного поставщика событий.

В следующем примере показана функция C#, которая считывает и регистрирует сообщение Kafka как событие Kafka.

[FunctionName("KafkaTrigger")]
public static void Run(
    [KafkaTrigger("BrokerList",
                  "topic",
                  Username = "ConfluentCloudUserName",
                  Password = "ConfluentCloudPassword",
                  Protocol = BrokerProtocol.SaslSsl,
                  AuthenticationMode = BrokerAuthenticationMode.Plain,
                  ConsumerGroup = "$Default")] KafkaEventData<string> kevent, ILogger log)
{            
    log.LogInformation($"C# Kafka trigger function processed a message: {kevent.Value}");
}

Чтобы получить события в пакете, используйте входную строку или KafkaEventData в качестве массива, как показано в следующем примере:

[FunctionName("KafkaTriggerMany")]
public static void Run(
    [KafkaTrigger("BrokerList",
                  "topic",
                  Username = "ConfluentCloudUserName",
                  Password = "ConfluentCloudPassword",
                  Protocol = BrokerProtocol.SaslSsl,
                  AuthenticationMode = BrokerAuthenticationMode.Plain,
                  ConsumerGroup = "$Default")] KafkaEventData<string>[] events, ILogger log)
{       
    foreach (KafkaEventData<string> kevent in events)
    {    
        log.LogInformation($"C# Kafka trigger function processed a message: {kevent.Value}");
    }
}

Следующая функция регистрирует сообщение и заголовки для события Kafka.

[FunctionName("KafkaTriggerSingleWithHeaders")]
public static void Run(
    [KafkaTrigger("BrokerList",
                  "topic",
                  Username = "ConfluentCloudUserName",
                  Password = "ConfluentCloudPassword",
                  Protocol = BrokerProtocol.SaslSsl,
                  AuthenticationMode = BrokerAuthenticationMode.Plain,
                  ConsumerGroup = "$Default")] KafkaEventData<string> kevent, ILogger log)
{
    log.LogInformation($"C# Kafka trigger function processed a message: {kevent.Value}");
    log.LogInformation("Headers: ");
    var headers = kevent.Headers;
    foreach (var header in headers)
    {
        log.LogInformation($"Key = {header.Key} Value = {System.Text.Encoding.UTF8.GetString(header.Value)}");
    }
}

Вы можете определить универсальную схему Avro для события, переданного триггеру. Следующее строковое значение определяет универсальную схему Avro.

      const string PageViewsSchema = @"{
  ""type"": ""record"",
  ""name"": ""pageviews"",
  ""namespace"": ""ksql"",
  ""fields"": [
    {
      ""name"": ""viewtime"",
      ""type"": ""long""
    },
    {
      ""name"": ""userid"",
      ""type"": ""string""
    },
    {
      ""name"": ""pageid"",
      ""type"": ""string""
    }
  ]
}";

В следующей функции экземпляр GenericRecord доступен в свойстве KafkaEvent.Value.

[FunctionName(nameof(PageViews))]
public static void PageViews(
   [KafkaTrigger("LocalBroker", "pageviews", AvroSchema = PageViewsSchema, ConsumerGroup = "azfunc")] KafkaEventData<string, GenericRecord>[] kafkaEvents,
   long[] offsetArray,
   int[] partitionArray,
   string[] topicArray,
   DateTime[] timestampArray,
   ILogger logger)
{
    for (int i = 0; i < kafkaEvents.Length; i++)
    {
        var kafkaEvent = kafkaEvents[i];
        if (kafkaEvent.Value is GenericRecord genericRecord)
        {
            logger.LogInformation($"[{timestampArray[i]}] {topicArray[i]} / {partitionArray[i]} / {offsetArray[i]}: {GenericToJson(genericRecord)}");
        }
    }
}

Вы можете определить конкретную схему Avro для события, переданного триггеру. В указанном ниже коде определяется класс UserRecord:

       public const string SchemaText = @"
       {
  ""type"": ""record"",
  ""name"": ""UserRecord"",
  ""namespace"": ""KafkaFunctionSample"",
  ""fields"": [
    {
      ""name"": ""registertime"",
      ""type"": ""long""
    },
    {
      ""name"": ""userid"",
      ""type"": ""string""
    },
    {
      ""name"": ""regionid"",
      ""type"": ""string""
    },
    {
      ""name"": ""gender"",
      ""type"": ""string""
    }
  ]
}";

В следующей функции экземпляр UserRecord доступен в свойстве KafkaEvent.Value.

[FunctionName(nameof(User))]
public static void User(
   [KafkaTrigger("LocalBroker", "users", ConsumerGroup = "azfunc")] KafkaEventData<string, UserRecord>[] kafkaEvents,
   ILogger logger)
{
    foreach (var kafkaEvent in kafkaEvents)
    {
        logger.LogInformation($"{JsonConvert.SerializeObject(kafkaEvent.Value)}");
    }
}

Полный набор рабочих примеров .NET см. в репозитории расширений Kafka.

В следующем примере показана функция C#, которая считывает и регистрирует сообщение Kafka как событие Kafka.

[FunctionName("KafkaTrigger")]
public static void Run(
    [KafkaTrigger("BrokerList",
                  "topic",
                  Username = "$ConnectionString",
                  Password = "%EventHubConnectionString%",
                  Protocol = BrokerProtocol.SaslSsl,
                  AuthenticationMode = BrokerAuthenticationMode.Plain,
                  ConsumerGroup = "$Default")] KafkaEventData<string> kevent, ILogger log)
{            
    log.LogInformation($"C# Kafka trigger function processed a message: {kevent.Value}");
}

Чтобы получить события в пакете, используйте в качестве входных данных массив строк или массив KafkaEventData, как показано в следующем примере:

[FunctionName("KafkaTriggerMany")]
public static void Run(
    [KafkaTrigger("BrokerList",
                  "topic",
                  Username = "$ConnectionString",
                  Password = "%EventHubConnectionString%",
                  Protocol = BrokerProtocol.SaslSsl,
                  AuthenticationMode = BrokerAuthenticationMode.Plain,
                  ConsumerGroup = "$Default")] KafkaEventData<string>[] events, ILogger log)
{       
    foreach (KafkaEventData<string> kevent in events)
    {    
        log.LogInformation($"C# Kafka trigger function processed a message: {kevent.Value}");
    }
}

Следующая функция регистрирует сообщение и заголовки для события Kafka.

[FunctionName("KafkaTriggerSingleWithHeaders")]
public static void Run(
    [KafkaTrigger("BrokerList",
                  "topic",
                  Username = "$ConnectionString",
                  Password = "%EventHubConnectionString%",
                  Protocol = BrokerProtocol.SaslSsl,
                  AuthenticationMode = BrokerAuthenticationMode.Plain,
                  ConsumerGroup = "$Default")] KafkaEventData<string> kevent, ILogger log)
{
    log.LogInformation($"C# Kafka trigger function processed a message: {kevent.Value}");
    log.LogInformation("Headers: ");
    var headers = kevent.Headers;
    foreach (var header in headers)
    {
        log.LogInformation($"Key = {header.Key} Value = {System.Text.Encoding.UTF8.GetString(header.Value)}");
    }

      const string PageViewsSchema = @"{
  ""type"": ""record"",
  ""name"": ""pageviews"",
  ""namespace"": ""ksql"",
  ""fields"": [
    {
      ""name"": ""viewtime"",
      ""type"": ""long""
    },
    {
      ""name"": ""userid"",
      ""type"": ""string""
    },
    {
      ""name"": ""pageid"",
      ""type"": ""string""
    }
  ]
}";

В следующей функции экземпляр GenericRecord доступен в свойстве KafkaEvent.Value.

[FunctionName(nameof(PageViews))]
public static void PageViews(
   [KafkaTrigger("LocalBroker", "pageviews", AvroSchema = PageViewsSchema, ConsumerGroup = "azfunc")] KafkaEventData<string, GenericRecord>[] kafkaEvents,
   long[] offsetArray,
   int[] partitionArray,
   string[] topicArray,
   DateTime[] timestampArray,
   ILogger logger)
{
    for (int i = 0; i < kafkaEvents.Length; i++)
    {
        var kafkaEvent = kafkaEvents[i];
        if (kafkaEvent.Value is GenericRecord genericRecord)
        {
            logger.LogInformation($"[{timestampArray[i]}] {topicArray[i]} / {partitionArray[i]} / {offsetArray[i]}: {GenericToJson(genericRecord)}");
        }
    }
}

       public const string SchemaText = @"
       {
  ""type"": ""record"",
  ""name"": ""UserRecord"",
  ""namespace"": ""KafkaFunctionSample"",
  ""fields"": [
    {
      ""name"": ""registertime"",
      ""type"": ""long""
    },
    {
      ""name"": ""userid"",
      ""type"": ""string""
    },
    {
      ""name"": ""regionid"",
      ""type"": ""string""
    },
    {
      ""name"": ""gender"",
      ""type"": ""string""
    }
  ]
}";

В следующей функции экземпляр UserRecord доступен в свойстве KafkaEvent.Value.

[FunctionName(nameof(User))]
public static void User(
   [KafkaTrigger("LocalBroker", "users", ConsumerGroup = "azfunc")] KafkaEventData<string, UserRecord>[] kafkaEvents,
   ILogger logger)
{
    foreach (var kafkaEvent in kafkaEvents)
    {
        logger.LogInformation($"{JsonConvert.SerializeObject(kafkaEvent.Value)}");
    }
}

Полный набор рабочих примеров .NET см. в репозитории расширений Kafka.

В следующем примере показана функция C#, которая считывает и регистрирует сообщение Kafka как событие Kafka.

[Function("KafkaTrigger")]
public static void Run(
    [KafkaTrigger("BrokerList",
                  "topic",
                  Username = "ConfluentCloudUserName",
                  Password = "ConfluentCloudPassword",
                  Protocol = BrokerProtocol.SaslSsl,
                  AuthenticationMode = BrokerAuthenticationMode.Plain,
                  ConsumerGroup = "$Default")] string eventData, FunctionContext context)
{
    var logger = context.GetLogger("KafkaFunction");
    logger.LogInformation($"C# Kafka trigger function processed a message: {JObject.Parse(eventData)["Value"]}");
}

Чтобы получить события в пакете, используйте в качестве входных данных массив строк, как показано в следующем примере:

[Function("KafkaTriggerMany")]
public static void Run(
    [KafkaTrigger("BrokerList",
                  "topic",
                  Username = "ConfluentCloudUserName",
                  Password = "ConfluentCloudPassword",
                  Protocol = BrokerProtocol.SaslSsl,
                  AuthenticationMode = BrokerAuthenticationMode.Plain,
                  ConsumerGroup = "$Default",
                  IsBatched = true)] string[] events, FunctionContext context)
{
    foreach (var kevent in events)
    {
        var logger = context.GetLogger("KafkaFunction");
        logger.LogInformation($"C# Kafka trigger function processed a message: {JObject.Parse(kevent)["Value"]}");
    }

Следующая функция регистрирует сообщение и заголовки для события Kafka.

[Function("KafkaTriggerWithHeaders")]
public static void Run(
    [KafkaTrigger("BrokerList",
                  "topic",
                  Username = "ConfluentCloudUserName",
                  Password = "ConfluentCloudPassword",
                  Protocol = BrokerProtocol.SaslSsl,
                  AuthenticationMode = BrokerAuthenticationMode.Plain,
                  ConsumerGroup = "$Default")] string eventData, FunctionContext context)
{
    var eventJsonObject = JObject.Parse(eventData);
    var logger = context.GetLogger("KafkaFunction");
    logger.LogInformation($"C# Kafka trigger function processed a message: {eventJsonObject["Value"]}");
    var headersJArr = eventJsonObject["Headers"] as JArray;
    logger.LogInformation("Headers for this event: ");
    foreach (JObject header in headersJArr)
    {
        logger.LogInformation($"{header["Key"]} {System.Text.Encoding.UTF8.GetString((byte[])header["Value"])}");

    }
}

Полный набор рабочих примеров .NET см. в репозитории расширений Kafka.

В следующем примере показана функция C#, которая считывает и регистрирует сообщение Kafka как событие Kafka.

[Function("KafkaTrigger")]
public static void Run(
    [KafkaTrigger("BrokerList",
                  "topic",
                  Username = "$ConnectionString",
                  Password = "EventHubConnectionString",
                  Protocol = BrokerProtocol.SaslSsl,
                  AuthenticationMode = BrokerAuthenticationMode.Plain,
                  ConsumerGroup = "$Default")] string eventData, FunctionContext context)
{
    var logger = context.GetLogger("KafkaFunction");
    logger.LogInformation($"C# Kafka trigger function processed a message: {JObject.Parse(eventData)["Value"]}");
}

[Function("KafkaTriggerMany")]
public static void Run(
    [KafkaTrigger("BrokerList",
                  "topic",
                  Username = "$ConnectionString",
                  Password = "EventHubConnectionString",
                  Protocol = BrokerProtocol.SaslSsl,
                  AuthenticationMode = BrokerAuthenticationMode.Plain,
                  ConsumerGroup = "$Default",
                  IsBatched = true)] string[] events, FunctionContext context)
{
    foreach (var kevent in events)
    {
        var logger = context.GetLogger("KafkaFunction");
        logger.LogInformation($"C# Kafka trigger function processed a message: {JObject.Parse(kevent)["Value"]}");
    }

Следующая функция регистрирует сообщение и заголовки для события Kafka.

[Function("KafkaTriggerWithHeaders")]
public static void Run(
    [KafkaTrigger("BrokerList",
                  "topic",
                  Username = "$ConnectionString",
                  Password = "EventHubConnectionString",
                  Protocol = BrokerProtocol.SaslSsl,
                  AuthenticationMode = BrokerAuthenticationMode.Plain,
                  ConsumerGroup = "$Default")] string eventData, FunctionContext context)
{
    var eventJsonObject = JObject.Parse(eventData);
    var logger = context.GetLogger("KafkaFunction");
    logger.LogInformation($"C# Kafka trigger function processed a message: {eventJsonObject["Value"]}");
    var headersJArr = eventJsonObject["Headers"] as JArray;
    logger.LogInformation("Headers for this event: ");
    foreach (JObject header in headersJArr)
    {
        logger.LogInformation($"{header["Key"]} {System.Text.Encoding.UTF8.GetString((byte[])header["Value"])}");

    }
}

Полный набор рабочих примеров .NET см. в репозитории расширений Kafka.

Примечание

Эквивалентный набор примеров TypeScript см. в репозитории расширений Kafka

Конкретные свойства файла function.json зависят от поставщика событий. В данных примерах это Confluent или Центры событий Azure. В следующих примерах показан триггер Kafka для функции, которая считывает и регистрирует сообщение Kafka.

Следующий файл function.json определяет триггер для конкретного поставщика.

Confluent
Центры событий

{
    "bindings": [
        {
            "type": "kafkaTrigger",
            "name": "event",
            "direction": "in",
            "topic": "topic",
            "brokerList": "%BrokerList%",
            "username": "%ConfluentCloudUserName%",
            "password": "%ConfluentCloudPassword%",
            "protocol": "saslSsl",
            "authenticationMode": "plain",
            "consumerGroup" : "$Default",
            "dataType": "string"
        }
    ]
}

{
    "bindings": [
        {
            "type": "kafkaTrigger",
            "name": "event",
            "direction": "in",
            "topic": "topic",
            "brokerList": "%BrokerList%",
            "username": "$ConnectionString",
            "password": "EventHubConnectionString",
            "protocol": "saslSsl",
            "authenticationMode": "plain",
            "consumerGroup" : "$Default",
            "dataType": "string"
        }
    ]
}

Затем при активации функции выполняется следующий код:

module.exports = async function (context, event) {
    // context.log.info(event)
    context.log.info(`JavaScript Kafka trigger function called for message ${event.Value}`);
};

Чтобы получить события в пакете, задайте для параметра cardinality в файле function.json значение many, как показано в приведенных ниже примерах.

Confluent
Центры событий

{
    "bindings": [
        {
            "type": "kafkaTrigger",
            "name": "event",
            "direction": "in",
            "protocol" : "SASLSSL",
            "password" : "%ConfluentCloudPassword%",
            "dataType" : "string",
            "topic" : "topic",
            "authenticationMode" : "PLAIN",
            "cardinality" : "MANY",
            "consumerGroup" : "$Default",
            "username" : "%ConfluentCloudUserName%",
            "brokerList" : "%BrokerList%"
        }
    ]
}

{
    "bindings": [
        {
            "type": "kafkaTrigger",
            "name": "events",
            "direction": "in",
            "protocol" : "SASLSSL",
            "dataType" : "string",
            "topic" : "topic",
            "authenticationMode" : "PLAIN",
            "cardinality" : "MANY",
            "consumerGroup" : "$Default",
            "username": "$ConnectionString",
            "password": "EventHubConnectionString",
            "brokerList" : "%BrokerList%"
        }
    ]
}

Затем следующий код анализирует массив событий и регистрирует данные события:

module.exports = async function (context, events) {
    function print(event) {
        var eventJson = JSON.parse(event)
        context.log.info(`JavaScript Kafka trigger function called for message ${eventJson.Value}`);
    }
    events.map(print);
};

Следующий код также регистрирует данные заголовка:

module.exports = async function (context, event) {
  function print(kevent) {
    var keventJson = JSON.parse(kevent)
    context.log.info(`JavaScript Kafka trigger function called for message ${keventJson.Value}`);
    context.log.info(`Headers for this message:`)
    let headers =  keventJson.Headers;
    headers.forEach(element => {
        context.log.info(`Key: ${element.Key} Value:${Buffer.from(element.Value, 'base64')}`) 
    });
  }
  event.map(print);
};

Вы можете определить универсальную схему Avro для события, переданного триггеру. Следующий файл function.json определяет триггер для конкретного поставщика, используя универсальную схему Avro.

Confluent
Центры событий

{
  "bindings" : [ {
    "type" : "kafkaTrigger",
    "direction" : "in",
    "name" : "kafkaAvroGenericSingle",
    "protocol" : "SASLSSL",
    "password" : "ConfluentCloudPassword",
    "topic" : "topic",
    "avroSchema" : "{\"type\":\"record\",\"name\":\"Payment\",\"namespace\":\"io.confluent.examples.clients.basicavro\",\"fields\":[{\"name\":\"id\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"double\"},{\"name\":\"type\",\"type\":\"string\"}]}",
    "authenticationMode" : "PLAIN",
    "consumerGroup" : "$Default",
    "username" : "ConfluentCloudUsername",
    "brokerList" : "%BrokerList%"
  } ]
}

{
  "bindings" : [ {
    "type" : "kafkaTrigger",
    "direction" : "in",
    "name" : "kafkaAvroGenericSingle",
    "protocol" : "SASLSSL",
    "password" : "EventHubConnectionString",
    "topic" : "topic",
    "authenticationMode" : "PLAIN",
    "avroSchema" : "{\"type\":\"record\",\"name\":\"Payment\",\"namespace\":\"io.confluent.examples.clients.basicavro\",\"fields\":[{\"name\":\"id\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"double\"},{\"name\":\"type\",\"type\":\"string\"}]}",
    "consumerGroup" : "$Default",
    "username" : "$ConnectionString",
    "brokerList" : "%BrokerList%"
  } ]
}

Затем при активации функции выполняется следующий код:

module.exports = async function (context, event) {
    context.log.info(`JavaScript Kafka trigger function called for message ${JSON.stringify(event)}`);
};

Полный набор рабочих примеров JavaScript см. в репозитории расширений Kafka.

Следующий файл function.json определяет триггер для конкретного поставщика.

Confluent
Центры событий

{
    "bindings": [
      {
            "type": "kafkaTrigger",
            "name": "kafkaEvent",
            "direction": "in",
            "protocol" : "SASLSSL",
            "password" : "%ConfluentCloudPassword%",
            "dataType" : "string",
            "topic" : "topic",
            "authenticationMode" : "PLAIN",
            "consumerGroup" : "$Default",
            "username" : "%ConfluentCloudUserName%",
            "brokerList" : "%BrokerList%",
            "sslCaLocation": "confluent_cloud_cacert.pem"
        }
    ]
}

{
    "bindings": [
      {
            "type": "kafkaTrigger",
            "name": "kafkaEvent",
            "direction": "in",
            "protocol" : "SASLSSL",
            "password" : "EventHubConnectionString",
            "dataType" : "string",
            "topic" : "topic",
            "authenticationMode" : "PLAIN",
            "consumerGroup" : "$Default",
            "username" : "$ConnectionString",
            "brokerList" : "%BrokerList%",
            "sslCaLocation": "confluent_cloud_cacert.pem"
        }
    ]
}

Затем при активации функции выполняется следующий код:

using namespace System.Net

param($kafkaEvent, $TriggerMetadata)

Write-Output "Powershell Kafka trigger function called for message $kafkaEvent.Value"

Confluent
Центры событий

{
    "bindings": [
      {
            "type": "kafkaTrigger",
            "name": "kafkaEvent",
            "direction": "in",
            "protocol" : "SASLSSL",
            "password" : "%ConfluentCloudPassword%",
            "dataType" : "string",
            "topic" : "topic",
            "authenticationMode" : "PLAIN",
            "cardinality" : "MANY",
            "consumerGroup" : "$Default",
            "username" : "%ConfluentCloudUserName%",
            "brokerList" : "%BrokerList%",
            "sslCaLocation": "confluent_cloud_cacert.pem"
        }
    ]
}

{
    "bindings": [
      {
            "type": "kafkaTrigger",
            "name": "kafkaEvents",
            "direction": "in",
            "protocol" : "SASLSSL",
            "password" : "EventHubConnectionString",
            "dataType" : "string",
            "topic" : "topic",
            "authenticationMode" : "PLAIN",
            "cardinality" : "MANY",
            "consumerGroup" : "$Default",
            "username" : "$ConnectionString",
            "brokerList" : "%BrokerList%"
        }
    ]
}

Затем следующий код анализирует массив событий и регистрирует данные события:

using namespace System.Net

param($kafkaEvents, $TriggerMetadata)

$kafkaEvents
foreach ($kafkaEvent in $kafkaEvents) {
    $event = $kafkaEvent | ConvertFrom-Json -AsHashtable
    Write-Output "Powershell Kafka trigger function called for message $event.Value"
}

Следующий код также регистрирует данные заголовка:

using namespace System.Net

param($kafkaEvents, $TriggerMetadata)

foreach ($kafkaEvent in $kafkaEvents) {
    $kevent = $kafkaEvent | ConvertFrom-Json -AsHashtable
    Write-Output "Powershell Kafka trigger function called for message $kevent.Value"
    Write-Output "Headers for this message:"
    foreach ($header in $kevent.Headers) {
        $DecodedValue = [System.Text.Encoding]::Unicode.GetString([System.Convert]::FromBase64String($header.Value))
        $Key = $header.Key
        Write-Output "Key: $Key Value: $DecodedValue"
    }
}

Confluent
Центры событий

{
  "bindings" : [ {
    "type" : "kafkaTrigger",
    "direction" : "in",
    "name" : "kafkaEvent",
    "protocol" : "SASLSSL",
    "password" : "ConfluentCloudPassword",
    "topic" : "topic",
    "authenticationMode" : "PLAIN",
    "avroSchema" : "{\"type\":\"record\",\"name\":\"Payment\",\"namespace\":\"io.confluent.examples.clients.basicavro\",\"fields\":[{\"name\":\"id\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"double\"},{\"name\":\"type\",\"type\":\"string\"}]}",
    "consumerGroup" : "$Default",
    "username" : "ConfluentCloudUsername",
    "brokerList" : "%BrokerList%"
  } ]
}

{
  "bindings" : [ {
    "type" : "kafkaTrigger",
    "direction" : "in",
    "name" : "kafkaTriggerAvroGeneric",
    "protocol" : "SASLSSL",
    "password" : "EventHubConnectionString",
    "topic" : "topic",
    "authenticationMode" : "PLAIN",
    "avroSchema" : "{\"type\":\"record\",\"name\":\"Payment\",\"namespace\":\"io.confluent.examples.clients.basicavro\",\"fields\":[{\"name\":\"id\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"double\"},{\"name\":\"type\",\"type\":\"string\"}]}",
    "consumerGroup" : "$Default",
    "username" : "$ConnectionString",
    "BrokerList" : "%BrokerList%"
  } ]
}

Затем при активации функции выполняется следующий код:

using namespace System.Net

param($kafkaEvent, $TriggerMetadata)

Write-Output "Powershell Kafka trigger function called for message $kafkaEvent.Value"

Полный набор рабочих примеров PowerShell см. в репозитории расширений Kafka.

Следующий файл function.json определяет триггер для конкретного поставщика.

Confluent
Центры событий

{
      "scriptFile": "main.py",
      "bindings": [
        {
          "type": "kafkaTrigger",
          "name": "kevent",
          "topic": "topic",
          "brokerList": "%BrokerList%",
          "username": "%ConfluentCloudUserName%",
          "password": "%ConfluentCloudPassword%",
          "consumerGroup" : "functions",
          "protocol": "saslSsl",
          "authenticationMode": "plain"
        }
    ]
}

{
      "scriptFile": "main.py",
      "bindings": [
        {
          "type": "kafkaTrigger",
          "name": "kevent",
          "topic": "topic",
          "brokerList": "%BrokerList%",
          "username": "$ConnectionString",
          "password": "EventHubConnectionString",
          "consumerGroup" : "functions",
          "protocol": "saslSsl",
          "authenticationMode": "plain"
        }
    ]
}

Затем при активации функции выполняется следующий код:

import logging
from azure.functions import KafkaEvent

def main(kevent : KafkaEvent):
    logging.info(kevent.get_body().decode('utf-8'))
    logging.info(kevent.metadata)

Confluent
Центры событий

{
      "scriptFile": "main.py",
      "bindings": [
        {
            "type" : "kafkaTrigger",
            "direction": "in",
            "name" : "kevents",
            "protocol" : "SASLSSL",
            "password" : "%ConfluentCloudPassword%",
            "topic" : "message_python",
            "authenticationMode" : "PLAIN",
            "cardinality" : "MANY",
            "dataType": "string",
            "consumerGroup" : "$Default",
            "username" : "%ConfluentCloudUserName%",
            "BrokerList" : "%BrokerList%"    
        }
    ]
}

{
      "scriptFile": "main.py",
      "bindings": [
        {
            "type" : "kafkaTrigger",
            "direction": "in",
            "name" : "kevents",
            "protocol" : "SASLSSL",
            "password" : "EventHubConnectionString",
            "topic" : "message_python",
            "authenticationMode" : "PLAIN",
            "cardinality" : "MANY",
            "dataType": "string",
            "consumerGroup" : "$Default",
            "username" : "$ConnectionString",
            "BrokerList" : "%BrokerList%"    
        }
    ]
}

Затем следующий код анализирует массив событий и регистрирует данные события:

import logging
import typing
from azure.functions import KafkaEvent

def main(kevents : typing.List[KafkaEvent]):
    for event in kevents:
        logging.info(event.get_body())

Следующий код также регистрирует данные заголовка:

import logging
import typing
from azure.functions import KafkaEvent
import json
import base64

def main(kevents : typing.List[KafkaEvent]):
    for event in kevents:
        event_dec = event.get_body().decode('utf-8')
        event_json = json.loads(event_dec)
        logging.info("Python Kafka trigger function called for message " + event_json["Value"])
        headers = event_json["Headers"]
        for header in headers:
            logging.info("Key: "+ header['Key'] + " Value: "+ str(base64.b64decode(header['Value']).decode('ascii')))

Confluent
Центры событий

{
  "scriptFile": "main.py",
  "bindings" : [ {
    "type" : "kafkaTrigger",
    "direction" : "in",
    "name" : "kafkaTriggerAvroGeneric",
    "protocol" : "SASLSSL",
    "password" : "ConfluentCloudPassword",
    "topic" : "topic",
    "authenticationMode" : "PLAIN",
    "avroSchema" : "{\"type\":\"record\",\"name\":\"Payment\",\"namespace\":\"io.confluent.examples.clients.basicavro\",\"fields\":[{\"name\":\"id\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"double\"},{\"name\":\"type\",\"type\":\"string\"}]}",
    "consumerGroup" : "$Default",
    "username" : "ConfluentCloudUsername",
    "brokerList" : "%BrokerList%"
  } ]
}

{
  "scriptFile": "main.py",
  "bindings" : [ {
    "type" : "kafkaTrigger",
    "direction" : "in",
    "name" : "kafkaTriggerAvroGeneric",
    "protocol" : "SASLSSL",
    "password" : "EventHubConnectionString",
    "topic" : "topic",
    "authenticationMode" : "PLAIN",
    "avroSchema" : "{\"type\":\"record\",\"name\":\"Payment\",\"namespace\":\"io.confluent.examples.clients.basicavro\",\"fields\":[{\"name\":\"id\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"double\"},{\"name\":\"type\",\"type\":\"string\"}]}",
    "consumerGroup" : "$Default",
    "username" : "$ConnectionString",
    "BrokerList" : "%BrokerList%"
  } ]
}

Затем при активации функции выполняется следующий код:

import logging
from azure.functions import KafkaEvent

def main(kafkaTriggerAvroGeneric : KafkaEvent):
    logging.info(kafkaTriggerAvroGeneric.get_body().decode('utf-8'))
    logging.info(kafkaTriggerAvroGeneric.metadata)

Полный набор рабочих примеров Python см. в репозитории расширений Kafka.

Заметки, используемые для настройки триггера, зависят от конкретного поставщика событий.

Confluent
Центры событий

В следующем примере показана функция Java, которая считывает и регистрирует содержимое события Kafka.

@FunctionName("KafkaTrigger")
public void runSingle(
        @KafkaTrigger(
            name = "KafkaTrigger",
            topic = "topic",  
            brokerList="%BrokerList%",
            consumerGroup="$Default", 
            username = "%ConfluentCloudUsername%", 
            password = "ConfluentCloudPassword",
            authenticationMode = BrokerAuthenticationMode.PLAIN,
            protocol = BrokerProtocol.SASLSSL,
            // sslCaLocation = "confluent_cloud_cacert.pem", // Enable this line for windows.
            dataType = "string"
         ) String kafkaEventData,
        final ExecutionContext context) {
        context.getLogger().info(kafkaEventData);
}

Чтобы получить события в пакете, используйте входную строку в качестве массива, как показано в следующем примере:

@FunctionName("KafkaTriggerMany")
public void runMany(
        @KafkaTrigger(
            name = "kafkaTriggerMany",
            topic = "topic",  
            brokerList="%BrokerList%",
            consumerGroup="$Default", 
            username = "%ConfluentCloudUsername%", 
            password = "ConfluentCloudPassword",
            authenticationMode = BrokerAuthenticationMode.PLAIN,
            protocol = BrokerProtocol.SASLSSL,
            // sslCaLocation = "confluent_cloud_cacert.pem", // Enable this line for windows.
            cardinality = Cardinality.MANY,
            dataType = "string"
         ) String[] kafkaEvents,
        final ExecutionContext context) {
        for (String kevent: kafkaEvents) {
            context.getLogger().info(kevent);
        }    
}

Следующая функция регистрирует сообщение и заголовки для события Kafka.

@FunctionName("KafkaTriggerManyWithHeaders")
public void runSingle(
        @KafkaTrigger(
            name = "KafkaTrigger",
            topic = "topic",  
            brokerList="%BrokerList%",
            consumerGroup="$Default", 
            username = "%ConfluentCloudUsername%", 
            password = "ConfluentCloudPassword",
            authenticationMode = BrokerAuthenticationMode.PLAIN,
            protocol = BrokerProtocol.SASLSSL,
            // sslCaLocation = "confluent_cloud_cacert.pem", // Enable this line for windows.
            dataType = "string",
            cardinality = Cardinality.MANY
         ) List<String> kafkaEvents,
        final ExecutionContext context) {
            Gson gson = new Gson(); 
            for (String keventstr: kafkaEvents) {
                KafkaEntity kevent = gson.fromJson(keventstr, KafkaEntity.class);
                context.getLogger().info("Java Kafka trigger function called for message: " + kevent.Value);
                context.getLogger().info("Headers for the message:");
                for (KafkaHeaders header : kevent.Headers) {
                    String decodedValue = new String(Base64.getDecoder().decode(header.Value));
                    context.getLogger().info("Key:" + header.Key + " Value:" + decodedValue);                    
                }                
            }
        }

Вы можете определить универсальную схему Avro для события, переданного триггеру. Следующая функция определяет триггер для конкретного поставщика, используя универсальную схему Avro.

private static final String schema = "{\"type\":\"record\",\"name\":\"Payment\",\"namespace\":\"io.confluent.examples.clients.basicavro\",\"fields\":[{\"name\":\"id\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"double\"},{\"name\":\"type\",\"type\":\"string\"}]}";

@FunctionName("KafkaAvroGenericTrigger")
public void runOne(
        @KafkaTrigger(
                name = "kafkaAvroGenericSingle",
                topic = "topic",
                brokerList="%BrokerList%",
                consumerGroup="$Default",
                username = "ConfluentCloudUsername",
                password = "ConfluentCloudPassword",
                avroSchema = schema,
                authenticationMode = BrokerAuthenticationMode.PLAIN,
                protocol = BrokerProtocol.SASLSSL) Payment payment,
        final ExecutionContext context) {
    context.getLogger().info(payment.toString());
}

Полный набор рабочих примеров Java для Confluent см. в репозитории расширений Kafka.

В следующем примере показана функция Java, которая считывает и регистрирует содержимое события Kafka.

@FunctionName("KafkaTrigger")
public void runSingle(
        @KafkaTrigger(
            name = "KafkaTrigger",
            topic = "topic",  
            brokerList="%BrokerList%",
            consumerGroup="$Default", 
            username = "$ConnectionString", 
            password = "EventHubConnectionString",
            authenticationMode = BrokerAuthenticationMode.PLAIN,
            protocol = BrokerProtocol.SASLSSL,
            // sslCaLocation = "confluent_cloud_cacert.pem", // Enable this line for windows.
            dataType = "string"
         ) String kafkaEventData,
        final ExecutionContext context) {
        context.getLogger().info(kafkaEventData);
}

@FunctionName("KafkaTriggerMany")
public void runMany(
        @KafkaTrigger(
            name = "kafkaTriggerMany",
            topic = "topic",  
            brokerList="%BrokerList%",
            consumerGroup="$Default", 
            username = "$ConnectionString", 
            password = "EventHubConnectionString",
            authenticationMode = BrokerAuthenticationMode.PLAIN,
            protocol = BrokerProtocol.SASLSSL,
            // sslCaLocation = "confluent_cloud_cacert.pem", // Enable this line for windows.
            cardinality = Cardinality.MANY,
            dataType = "string"
         ) String[] kafkaEvents,
        final ExecutionContext context) {
        for (String kevent: kafkaEvents) {
            context.getLogger().info(kevent);
        }    
}

Следующая функция регистрирует сообщение и заголовки для события Kafka.

            @KafkaTrigger(
                name = "KafkaTrigger",
                topic = "topic",  
                brokerList="%BrokerList%",
                consumerGroup="$Default", 
                username = "$ConnectionString", 
                password = "EventHubConnectionString",
                authenticationMode = BrokerAuthenticationMode.PLAIN,
                protocol = BrokerProtocol.SASLSSL,
                // sslCaLocation = "confluent_cloud_cacert.pem", // Enable this line for windows.
                dataType = "string",
                cardinality = Cardinality.MANY
             ) List<String> kafkaEvents,
            final ExecutionContext context) {
                Gson gson = new Gson(); 
                for (String keventstr: kafkaEvents) {
                    KafkaEntity kevent = gson.fromJson(keventstr, KafkaEntity.class);
                    context.getLogger().info("Java Kafka trigger function called for message: " + kevent.Value);
                    context.getLogger().info("Headers for the message:");
                    for (KafkaHeaders header : kevent.Headers) {
                        String decodedValue = new String(Base64.getDecoder().decode(header.Value));
                        context.getLogger().info("Key:" + header.Key + " Value:" + decodedValue);                    
                    }                
                }
            }     
}


private static final String schema = "{\"type\":\"record\",\"name\":\"Payment\",\"namespace\":\"io.confluent.examples.clients.basicavro\",\"fields\":[{\"name\":\"id\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"double\"},{\"name\":\"type\",\"type\":\"string\"}]}";

@FunctionName("KafkaAvroGenericTrigger")
public void runOne(
        @KafkaTrigger(
                name = "kafkaAvroGenericSingle",
                topic = "topic",
                brokerList="%BrokerList%",
                consumerGroup="$Default",
                username = "$ConnectionString",
                password = "EventHubConnectionString",
                avroSchema = schema,
                authenticationMode = BrokerAuthenticationMode.PLAIN,
                protocol = BrokerProtocol.SASLSSL) Payment payment,
        final ExecutionContext context) {
    context.getLogger().info(payment.toString());

Полный набор рабочих примеров Java для Центров событий см. в репозитории расширений Kafka.

Атрибуты

Библиотеки C# в процессе и изолированном рабочем процессе используются KafkaTriggerAttribute для определения триггера функции.

В следующей таблице описаны свойства, которые можно задать с помощью этого атрибута триггера:

Параметр	Описание
BrokerList	(Обязательно) Список брокеров Kafka, отслеживаемых триггером. Дополнительные сведения см. в разделе Подключения.
Раздел	(Обязательно) Раздел, отслеживаемый триггером.
ConsumerGroup	(Необязательно) Группа потребителей Kafka, используемая триггером.
AvroSchema	(Необязательно) Схема универсальной записи при использовании протокола Avro.
AuthenticationMode	(Необязательно) Режим, используемый при простой проверке подлинности и проверке подлинности уровня безопасности (SASL). Поддерживаемые значения: `Gssapi`, `Plain` (по умолчанию), `ScramSha256`, `ScramSha512`.
Username	(Необязательно) Имя пользователя для проверки подлинности SASL. Не поддерживается, если `AuthenticationMode` имеет значение `Gssapi`. Дополнительные сведения см. в разделе Подключения.
Пароль	(Необязательно) Пароль для проверки подлинности SASL. Не поддерживается, если `AuthenticationMode` имеет значение `Gssapi`. Дополнительные сведения см. в разделе Подключения.
Протокол	(Необязательно) Протокол безопасности, используемый при взаимодействии с брокерами. Поддерживаемые значения: `plaintext` (по умолчанию), `ssl`, `sasl_plaintext`, `sasl_ssl`.
SslCaLocation	(Необязательно) Путь к файлу сертификата ЦС для проверки сертификата брокера.
SslCertificateLocation	(Необязательно) Путь к сертификату клиента.
SslKeyLocation	(Необязательно) Путь к закрытому ключу клиента (PEM), используемому для проверки подлинности.
SslKeyPassword	(Необязательно) Пароль для сертификата клиента.

Заметки

Заметка KafkaTrigger позволяет создать функцию, которая выполняется при получении раздела. Поддерживаемые варианты включают следующие элементы:

Элемент	Description
name	(Обязательно) Имя переменной, представляющей сообщение очереди или раздела в коде функции.
brokerList	(Обязательно) Список брокеров Kafka, отслеживаемых триггером. Дополнительные сведения см. в разделе Подключения.
topic	(Обязательно) Раздел, отслеживаемый триггером.
кратность	(Необязательно) Указывает кратность входных данных триггера. Поддерживаемые значения: `ONE` (по умолчанию) и `MANY`. Используйте `ONE`, если входные данные являются одним сообщением, и `MANY`, если входные данные являются массивом сообщений. При использовании `MANY` необходимо также задать параметр `dataType`.
dataType	Определяет, как Функции обрабатывают значение параметра. По умолчанию полученное значение представляет собой строку, и Функции пытаются десериализовать ее до объекта POJO. При `string` входные данные обрабатываются просто как строка. При `binary` сообщение приходит в формате двоичных данных и Функции пытаются десериализовать его до фактического типа параметра byte[].
consumerGroup	(Необязательно) Группа потребителей Kafka, используемая триггером.
avroSchema	(Необязательно) Схема универсальной записи при использовании протокола Avro.
authenticationMode	(Необязательно) Режим, используемый при простой проверке подлинности и проверке подлинности уровня безопасности (SASL). Поддерживаемые значения: `Gssapi`, `Plain` (по умолчанию), `ScramSha256`, `ScramSha512`.
username	(Необязательно) Имя пользователя для проверки подлинности SASL. Не поддерживается, если `AuthenticationMode` имеет значение `Gssapi`. Дополнительные сведения см. в разделе Подключения.
пароль	(Необязательно) Пароль для проверки подлинности SASL. Не поддерживается, если `AuthenticationMode` имеет значение `Gssapi`. Дополнительные сведения см. в разделе Подключения.
protocol	(Необязательно) Протокол безопасности, используемый при взаимодействии с брокерами. Поддерживаемые значения: `plaintext` (по умолчанию), `ssl`, `sasl_plaintext`, `sasl_ssl`.
sslCaLocation	(Необязательно) Путь к файлу сертификата ЦС для проверки сертификата брокера.
sslCertificateLocation	(Необязательно) Путь к сертификату клиента.
sslKeyLocation	(Необязательно) Путь к закрытому ключу клиента (PEM), используемому для проверки подлинности.
sslKeyPassword	(Необязательно) Пароль для сертификата клиента.

Настройка

В следующей таблице описываются свойства конфигурации привязки, которые задаются в файле function.json.

Свойство в function.json	Описание
type	Обязательное. Необходимо задать значение `kafkaTrigger`.
direction	Обязательное. Необходимо задать значение `in`.
name	(Обязательно) Имя переменной, представляющей данные, полученные через брокер, в коде функции.
brokerList	(Обязательно) Список брокеров Kafka, отслеживаемых триггером. Дополнительные сведения см. в разделе Подключения.
topic	(Обязательно) Раздел, отслеживаемый триггером.
кратность	(Необязательно) Указывает кратность входных данных триггера. Поддерживаемые значения: `ONE` (по умолчанию) и `MANY`. Используйте `ONE`, если входные данные являются одним сообщением, и `MANY`, если входные данные являются массивом сообщений. При использовании `MANY` необходимо также задать параметр `dataType`.
dataType	Определяет, как Функции обрабатывают значение параметра. По умолчанию полученное значение представляет собой строку, и Функции пытаются десериализовать ее до объекта POJO. При `string` входные данные обрабатываются просто как строка. При `binary` сообщение приходит в формате двоичных данных и Функции пытаются десериализовать его до фактического типа параметра byte[].
consumerGroup	(Необязательно) Группа потребителей Kafka, используемая триггером.
avroSchema	(Необязательно) Схема универсальной записи при использовании протокола Avro.
authenticationMode	(Необязательно) Режим, используемый при простой проверке подлинности и проверке подлинности уровня безопасности (SASL). Поддерживаемые значения: `Gssapi`, `Plain` (по умолчанию), `ScramSha256`, `ScramSha512`.
username	(Необязательно) Имя пользователя для проверки подлинности SASL. Не поддерживается, если `AuthenticationMode` имеет значение `Gssapi`. Дополнительные сведения см. в разделе Подключения.
пароль	(Необязательно) Пароль для проверки подлинности SASL. Не поддерживается, если `AuthenticationMode` имеет значение `Gssapi`. Дополнительные сведения см. в разделе Подключения.
protocol	(Необязательно) Протокол безопасности, используемый при взаимодействии с брокерами. Поддерживаемые значения: `plaintext` (по умолчанию), `ssl`, `sasl_plaintext`, `sasl_ssl`.
sslCaLocation	(Необязательно) Путь к файлу сертификата ЦС для проверки сертификата брокера.
sslCertificateLocation	(Необязательно) Путь к сертификату клиента.
sslKeyLocation	(Необязательно) Путь к закрытому ключу клиента (PEM), используемому для проверки подлинности.
sslKeyPassword	(Необязательно) Пароль для сертификата клиента.

События Kafka в настоящее время поддерживаются как строки и массивы строк, которые являются полезными данными JSON.

Сообщения Kafka передаются функции в виде строк и массивов строк, которые являются полезными данными JSON.

В плане "Премиум" необходимо включить мониторинг масштабирования среды выполнения для выходных данных Kafka, чтобы иметь возможность горизонтально увеличить масштаб до нескольких экземпляров. Дополнительные сведения см. в статье Включение масштабирования среды выполнения.

Для работы с триггерами Kafka нельзя использовать функцию "Тест и запуск" страницы "Код и тест" на портале Azure. Вместо этого необходимо отправлять тестовые события непосредственно в раздел, отслеживаемый триггером.

Полный набор поддерживаемых параметров host.json для триггера Kafka см. в статье Параметры host.json.

Связи

Все сведения о подключении, необходимые триггерам и привязкам, должны храниться в параметрах приложения, а не в определениях привязок в коде. Это верно для учетных данных, которые никогда не должны храниться в коде.

Важно!

Параметры учетных данных должны ссылаться на параметр приложения. Не прописывайте учетные данные в файлах кода или конфигурации. При локальном запуске используйте файл local.settings.json для учетных данных и не публикуйте его.

Confluent
Центры событий

При подключении к управляемому кластеру Kafka, предоставляемому Confluent в Azure, убедитесь, что в триггере или привязке заданы следующие учетные данные проверки подлинности для среды Confluent Cloud:

Параметр	Рекомендуемое значение	Description
BrokerList	`BootstrapServer`	Параметр приложения с именем `BootstrapServer` содержит значение сервера начальной загрузки, найденное на странице параметров Confluent Cloud. Значение напоминает `xyz-xyzxzy.westeurope.azure.confluent.cloud:9092`.
Username	`ConfluentCloudUsername`	Параметр приложения с именем `ConfluentCloudUsername` содержит ключ доступа API, полученный от веб-сайта Confluent Cloud.
Пароль	`ConfluentCloudPassword`	Параметр приложения с именем `ConfluentCloudPassword` содержит секрет API, полученный от веб-сайта Confluent Cloud.

При подключении к Центрам событий убедитесь, что в триггере или привязке заданы следующие учетные данные проверки подлинности для экземпляра Центров событий:

Параметр	Рекомендуемое значение	Description
BrokerList	`BootstrapServer`	Параметр приложения с именем `BootstrapServer` содержит полное доменное имя экземпляра Центров событий. Значение напоминает `<MY_NAMESPACE_NAME>.servicebus.windows.net:9093`.
Username	`$ConnectionString`	Фактическое значение предоставляется строкой подключения.
Пароль	`%EventHubsConnectionString%`	Параметр приложения с именем `EventHubsConnectionString` содержит строку подключения для пространства имен Центров событий. Дополнительные сведения см. в статье Получение строки подключения для Центров событий.

Строковые значения, используемые для этих параметров, должны присутствовать в качестве параметров приложения в Azure или коллекции Values в файле local.settings.json во время локальной разработки.

Необходимо также задать Protocol, AuthenticationMode и SslCaLocation в определениях привязок.

Следующие шаги

Запись в поток Apache Kafka из функции

Поделиться через

Триггер Apache Kafka для Функций Azure

Пример

Атрибуты

Заметки

Настройка

Использование

Связи

Следующие шаги

Обратная связь

Дополнительные ресурсы