Интеграция Функции Azure с Azure Data Explorer с помощью входных и выходных привязок (предварительная версия)

Функции Azure позволяют запускать бессерверный код в облаке по расписанию или в ответ на событие. Используя входные и выходные привязки Azure Data Explorer для Функций Azure, вы можете интегрировать Azure Data Explorer в рабочие процессы для приема данных и выполнения запросов к кластеру.

Предварительные условия

• Подписка Azure. Создайте бесплатную учетную запись Azure.
• Кластер и база данных Azure Data Explorer с примерами данных. Создайте кластер и базу данных.
• Учетная запись хранилища.

Попробуйте выполнить интеграцию с помощью примера проекта.

Использование привязок Azure Data Explorer для функции Azure

Сведения об использовании привязок Azure Data Explorer для функций Azure см. в следующих статьях:

• Обзор привязок Azure Data Explorer для Azure Functions
• Входные привязки Azure Data Explorer для Azure Functions
• Привязки для вывода данных Azure Data Explorer для Azure Functions

Сценарии использования привязок Azure Data Explorer для Функций Azure

В следующих разделах описаны некоторые распространенные сценарии использования привязок Azure Data Explorer для Функции Azure.

Входные привязки данных

Входные привязки выполняют запрос языка запросов Kusto (KQL) или функцию KQL, при необходимости с параметрами и возвращают выходные данные функции.

В следующих разделах описывается использование входных привязок в некоторых распространенных сценариях.

Сценарий 1. Конечная точка HTTP для запроса данных из кластера

Используйте входные привязки, если необходимо предоставить данные Azure Data Explorer через REST API. В этом сценарии используется HTTP-триггер Функции Azure для запроса данных в вашем кластере. Этот сценарий полезен в ситуациях, когда необходимо предоставить программный доступ к данным Azure Data Explorer для внешних приложений или служб. Предоставление данных через REST API позволяет приложениям легко использовать данные без необходимости подключаться непосредственно к кластеру.

Код определяет функцию с триггером HTTP и входной привязкой Azure Data Explorer. Входная привязка указывает запрос для выполнения в таблице Products в базе данных productsdb . Функция использует столбец productId в качестве предиката, передаваемого в качестве параметра.

{
    [FunctionName("GetProduct")]
    public static async Task<IActionResult> RunAsync(
        [HttpTrigger(AuthorizationLevel.User, "get", Route = "getproducts/{productId}")]
        HttpRequest req,
        [Kusto(Database:"productsdb" ,
        KqlCommand = "declare query_parameters (productId:long);Products | where ProductID == productId" ,
        KqlParameters = "@productId={productId}",
        Connection = "KustoConnectionString")]
        IAsyncEnumerable<Product> products)
    {
        IAsyncEnumerator<Product> enumerator = products.GetAsyncEnumerator();
        var productList = new List<Product>();
        while (await enumerator.MoveNextAsync())
        {
            productList.Add(enumerator.Current);
        }
        await enumerator.DisposeAsync();
        return new OkObjectResult(productList);
    }
}

Вы вызываете функцию следующим образом:

curl https://myfunctionapp.azurewebsites.net/api/getproducts/1

Сценарий 2. Запланированный триггер для экспорта данных из кластера

Следующий сценарий применим в ситуациях, когда данные необходимо экспортировать по расписанию, основанному на времени.

Код определяет функцию с триггером таймера, который экспортирует агрегирование данных о продажах из базы данных productsdb в CSV-файл в Хранилище BLOB-объектов Azure.

public static async Task Run([TimerTrigger("0 0 1 * * *")] TimerInfo myTimer,
    [Kusto(ConnectionStringSetting = "KustoConnectionString",
            DatabaseName = "productsdb",
            Query = "ProductSales | where OrderDate >= ago(1d) | summarize Sales = sum(ProductSales) by ProductName | top 10 by Sales desc")] IEnumerable<dynamic> queryResults,
[Blob("salescontainer/productsblob.csv", FileAccess.Write, Connection = "BlobStorageConnection")] CloudBlockBlob outputBlob,
ILogger log)
{
    // Write the query results to a CSV file
    using (var stream = new MemoryStream())
    using (var writer = new StreamWriter(stream))
    using (var csv = new CsvWriter(writer, CultureInfo.InvariantCulture))
    {
        csv.WriteRecords(queryResults);
        writer.Flush();
        stream.Position = 0;
        await outputBlob.UploadFromStreamAsync(stream);
    }
}

Выходные привязки

Выходные привязки принимают одну или несколько строк и вставляют их в таблицу Azure Data Explorer.

В следующих разделах описывается использование выходных привязок в некоторых распространенных сценариях.

Сценарий 1. Конечная точка HTTP для приема данных в кластер

Используйте этот сценарий, когда необходимо обработать входящие HTTP-запросы и принять данные в кластер. Используя выходную привязку, можно записывать входящие данные из запроса в таблицы Azure Data Explorer.

Код определяет функцию с триггером HTTP и выходной привязкой Azure Data Explorer. Эта функция принимает полезные данные JSON в тексте HTTP-запроса и записывает их в таблицу продуктов в базе данных productsdb.

public static IActionResult Run(
    [HttpTrigger(AuthorizationLevel.Anonymous, "post", Route = "addproductuni")]
    HttpRequest req, ILogger log,
    [Kusto(Database:"productsdb" ,
    TableName ="products" ,
    Connection = "KustoConnectionString")] out Product product)
{
    log.LogInformation($"AddProduct function started");
    string body = new StreamReader(req.Body).ReadToEnd();
    product = JsonConvert.DeserializeObject<Product>(body);
    string productString = string.Format(CultureInfo.InvariantCulture, "(Name:{0} ID:{1} Cost:{2})",
                product.Name, product.ProductID, product.Cost);
    log.LogInformation("Ingested product {}", productString);
    return new CreatedResult($"/api/addproductuni", product);
}

Вы вызываете функцию следующим образом:

curl -X POST https://myfunctionapp.azurewebsites.net/api/addproductuni -d '{"Name":"Product1","ProductID":1,"Cost":100,"ActivatedOn":"2023-01-02T00:00:00"}'

Сценарий 2. Прием данных из RabbitMQ или других систем обмена сообщениями, поддерживаемых в Azure

Используйте этот сценарий, когда необходимо принять данные из системы обмена сообщениями в кластере. Используя привязку вывода, вы можете загружать данные из системы обмена сообщениями в таблицы Azure Data Explorer.

Код определяет функцию с триггером RabbitMQ. Функция отправляет сообщения, данные в формате JSON в таблицу продуктов в базе данных productsdb .

public class QueueTrigger
{
    [FunctionName("QueueTriggerBinding")]
    [return: Kusto(Database: "productsdb",
                TableName = "products",
                Connection = "KustoConnectionString")]
    public static Product Run(
        [RabbitMQTrigger(queueName: "bindings.products.queue", ConnectionStringSetting = "rabbitMQConnectionAppSetting")] Product product,
        ILogger log)
    {
        log.LogInformation($"Dequeued product {product.ProductID}");
        return product;
    }
}

Дополнительные сведения о функциях см. документацию по Функциям Azure. Расширение Azure Data Explorer доступно в следующих разделах:

• GitHub
• NuGet.org

Связанный контент

• Обзор языка запросов Kusto (KQL)