Mengintegrasikan Azure Functions dengan Azure Data Explorer menggunakan pengikatan input dan output (pratinjau)

Penting

Konektor ini dapat digunakan dalam Real-Time Intelligence di Microsoft Fabric. Gunakan instruksi dalam artikel ini dengan pengecualian berikut:

  • Jika diperlukan, buat database menggunakan instruksi di Membuat database KQL.
  • Jika diperlukan, buat tabel menggunakan instruksi di Membuat tabel kosong.
  • Dapatkan URI kueri atau penyerapan menggunakan instruksi di Salin URI.
  • Jalankan kueri dalam set kueri KQL.

Azure Functions memungkinkan Anda menjalankan kode tanpa server di cloud sesuai jadwal atau menanggapi peristiwa. Dengan menggunakan pengikatan input dan output Azure Data Explorer untuk Azure Functions, Anda dapat mengintegrasikan Azure Data Explorer ke dalam alur kerja Anda untuk menyerap data dan menjalankan kueri terhadap kluster Anda.

Prasyarat

Cobalah integrasi dengan menggunakan proyek sampel.

Cara menggunakan pengikatan Azure Data Explorer untuk Azure Functions

Untuk informasi tentang cara menggunakan pengikatan Azure Data Explorer untuk Azure Functions, lihat artikel berikut ini:

Skenario untuk menggunakan pengikatan Azure Data Explorer untuk Azure Functions

Bagian berikut ini menjelaskan beberapa skenario umum untuk menggunakan pengikatan Azure Data Explorer untuk Azure Functions.

Pengikatan masukan

Pengikatan input menjalankan kueri Kusto Query Language (KQL) atau fungsi KQL, secara opsional dengan parameter, dan mengembalikan output ke fungsi .

Bagian berikut menjelaskan cara menggunakan pengikatan input dalam beberapa skenario umum.

Skenario 1: Titik akhir HTTP untuk mengkueri data dari kluster

Gunakan pengikatan input saat Anda perlu mengekspos data Azure Data Explorer melalui REST API. Dalam skenario ini, Anda menggunakan pemicu HTTP Azure Functions untuk mengkueri data di kluster Anda. Skenario ini berguna dalam situasi di mana Anda perlu menyediakan akses terprogram ke data Azure Data Explorer untuk aplikasi atau layanan eksternal. Mengekspos data Anda melalui REST API memungkinkan aplikasi untuk dengan mudah menggunakan data tanpa mengharuskannya untuk terhubung langsung ke kluster Anda.

Kode menentukan fungsi dengan pemicu HTTP dan pengikatan input Azure Data Explorer. Pengikatan input menentukan kueri yang akan dijalankan terhadap tabel Produk dalam database productsdb . Fungsi ini menggunakan kolom productId sebagai predikat yang diteruskan sebagai parameter.

{
    [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);
    }
}

Anda memanggil fungsi sebagai berikut:

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

Skenario 2: Pemicu terjadwal untuk mengekspor data dari kluster

Skenario berikut berlaku dalam situasi di mana data perlu diekspor pada jadwal berbasis waktu.

Kode menentukan fungsi dengan pemicu timer yang mengekspor agregasi data penjualan dari database productsdb ke file CSV di Azure Blob Storage.

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);
    }
}

Pengikatan output

Pengikatan output mengambil satu atau beberapa baris dan menyisipkannya ke dalam tabel Azure Data Explorer.

Bagian berikut menjelaskan cara menggunakan pengikatan output dalam beberapa skenario umum.

Skenario 1: Titik akhir HTTP untuk menyerap data ke dalam kluster

Gunakan skenario ini saat Anda perlu memproses permintaan HTTP masuk dan menyerap data ke dalam kluster Anda. Dengan menggunakan pengikatan output, Anda dapat menulis data masuk dari permintaan ke tabel Azure Data Explorer.

Kode menentukan fungsi dengan pemicu HTTP dan pengikatan output Azure Data Explorer. Fungsi ini mengambil payload JSON di isi permintaan HTTP dan menulisnya ke tabel produk dalam database 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);
}

Anda memanggil fungsi sebagai berikut:

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

Skenario 2: Menyerap data dari RabbitMQ atau sistem olahpesan lainnya yang didukung di Azure

Gunakan skenario ini saat Anda perlu menyerap data dari sistem olahpesan ke dalam kluster Anda. Dengan menggunakan pengikatan output, Anda dapat menyerap data masuk dari sistem olahpesan ke dalam tabel Azure Data Explorer.

Kode mendefinisikan fungsi dengan pemicu RabbitMQ. Fungsi ini menyerap pesan, data dalam format JSON, ke dalam tabel produk dalam database 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;
    }
}

Untuk informasi selengkapnya tentang fungsi, lihat dokumentasi Azure Functions. Ekstensi Azure Data Explorer tersedia di:

Konten terkait

  • Last updated on