Настройка приема потоковой передачи в пуле Data Explorer (предварительная версия) в Azure Synapse

При создании пула Data Explorer с помощью действий, описанных в разделе Создание пула Data Explorer, используя Synapse Studio, на вкладке Дополнительные параметры выберите Прием потоковой передачи>Включено.

При создании пула Data Explorer с помощью действий, описанных в разделе Создание пула Data Explorer, используя портал Azure, на вкладке Дополнительные параметры выберите Прием потоковой передачи>Включено.

1. В Synapse Studio в области слева выберите элемент Разработка.

2. В разделе Скрипты KQL выберите + (Добавить новый ресурс) >Скрипт KQL. В области справа можно присвоить имя вашему сценарию.

3. В меню Подключение к выберите contosodataexplorer.

4. В меню Использование базы данных выберите TestDatabase.

5. Вставьте следующую команду и выберите Выполнить, чтобы создать таблицу.

   .create table TestTable (TimeStamp: datetime, Name: string, Metric: int, Source:string)
   

6. Скопируйте одну из следующих команд в Панель запросов и выберите Выполнить. Будет определена политика приема потоковой передачи для созданной вами таблицы или базы данных, содержащей эту таблицу.

   Совет

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

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

     .alter table TestTable policy streamingingestion enable
     

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

     .alter database StreamingTestDb policy streamingingestion enable
     

1. На портале Azure Synapse Analytics перейдите к своему пулу Data Explorer.

2. Выберите Запрос.

3. Чтобы создать таблицу, которая будет получать данные посредством потоковой передачи, скопируйте следующую команду в Панель запросов и выберите Выполнить.

   .create table TestTable (TimeStamp: datetime, Name: string, Metric: int, Source:string)
   

4. Скопируйте одну из следующих команд в Панель запросов и выберите Выполнить. Будет определена политика приема потоковой передачи для созданной вами таблицы или базы данных, содержащей эту таблицу.

   Совет

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

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

     .alter table TestTable policy streamingingestion enable
     

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

     .alter database StreamingTestDb policy streamingingestion enable
     

using Kusto.Data;
using Kusto.Ingest;
using System.IO;
using Kusto.Data.Common;

namespace StreamingIngestion
{
    class Program
    {
        static void Main(string[] args)
        {
            string poolPath = "https://<Poolname>.<WorkspaceName>.kusto.windows.net";
            string appId = "<appId>";
            string appKey = "<appKey>";
            string appTenant = "<appTenant>";
            string dbName = "<dbName>";
            string tableName = "<tableName>";

            // Create Kusto connection string with App Authentication
            var csb =
                new KustoConnectionStringBuilder(poolPath)
                    .WithAadApplicationKeyAuthentication(
                        applicationClientId: appId,
                        applicationKey: appKey,
                        authority: appTenant
                    );

            // Create a disposable client that will execute the ingestion
            using (IKustoIngestClient client = KustoIngestFactory.CreateStreamingIngestClient(csb))
            {
                // Initialize client properties
                var ingestionProperties =
                    new KustoIngestionProperties(
                        databaseName: dbName,
                        tableName: tableName
                    );

                // Ingest from a compressed file
                var fileStream = File.Open("MyFile.gz", FileMode.Open);
                // Create source options
                var sourceOptions = new StreamSourceOptions()
                {
                    CompressionType = DataSourceCompressionType.GZip,
                };
                // Ingest from stream
                var status = client.IngestFromStreamAsync(fileStream, ingestionProperties, sourceOptions).GetAwaiter().GetResult();
            }
        }
    }
}

from Azure Synapse Analytics.kusto.data import KustoConnectionStringBuilder

from Azure Synapse Analytics.kusto.ingest import (
    IngestionProperties,
    DataFormat,
    KustoStreamingIngestClient
)

poolPath = "https://<Poolname>.<WorkspaceName>.kusto.windows.net"
appId = "<appId>"
appKey = "<appKey>"
appTenant = "<appTenant>"
dbName = "<dbName>"
tableName = "<tableName>"

csb = KustoConnectionStringBuilder.with_aad_application_key_authentication(
    poolPath,
    appId,
    appKey,
    appTenant
)
client = KustoStreamingIngestClient(csb)

ingestionProperties = IngestionProperties(
    database=dbName,
    table=tableName,
    data_format=DataFormat.CSV
)

# Ingest from file
# Automatically detects gz format
client.ingest_from_file("MyFile.gz", ingestion_properties=ingestionProperties) 

// Load modules using ES6 import statements:
import { DataFormat, IngestionProperties, StreamingIngestClient } from "azure-kusto-ingest";
import { KustoConnectionStringBuilder } from "azure-kusto-data";

// For earlier version, load modules using require statements:
// const IngestionProperties = require("azure-kusto-ingest").IngestionProperties;
// const KustoConnectionStringBuilder = require("azure-kusto-data").KustoConnectionStringBuilder;
// const {DataFormat} = require("azure-kusto-ingest").IngestionPropertiesEnums;
// const StreamingIngestClient = require("azure-kusto-ingest").StreamingIngestClient;

const poolPath = "https://<Poolname>.<WorkspaceName>.kusto.windows.net";
const appId = "<appId>";
const appKey = "<appKey>";
const appTenant = "<appTenant>";
const dbName = "<dbName>";
const tableName = "<tableName>";
const mappingName = "<mappingName>"; // Required for JSON formatted files

const ingestionProperties = new IngestionProperties({
    database: dbName, // Your database
    table: tableName, // Your table
    format: DataFormat.JSON,
    ingestionMappingReference: mappingName
});

// Initialize client with engine endpoint
const client = new StreamingIngestClient(
    KustoConnectionStringBuilder.withAadApplicationKeyAuthentication(
        poolPath,
        appId,
        appKey,
        appTenant
    ),
    ingestionProperties
);

// Automatically detects gz format
await client.ingestFromFile("MyFile.gz", ingestionProperties);

import (
    "context"
    "github.com/Azure Synapse Analytics/azure-kusto-go/kusto"
    "github.com/Azure Synapse Analytics/azure-kusto-go/kusto/ingest"
    "github.com/Azure Synapse Analytics/go-autorest/autorest/Azure Synapse Analytics/auth"
)

func ingest() {
    poolPath := "https://<Poolname>.<WorkspaceName>.kusto.windows.net"
    appId := "<appId>"
    appKey := "<appKey>"
    appTenant := "<appTenant>"
    dbName := "<dbName>"
    tableName := "<tableName>"
    mappingName := "<mappingName>" // Optional, can be nil

    // Creates a Kusto Authorizer using your client identity, secret, and tenant identity.
    // You may also uses other forms of authorization, see GoDoc > Authorization type.
    // auth package is: "github.com/Azure Synapse Analytics/go-autorest/autorest/Azure Synapse Analytics/auth"
    authorizer := kusto.Authorization{
        Config: auth.NewClientCredentialsConfig(appId, appKey, appTenant),
    }

    // Create a client
    client, err := kusto.New(poolPath, authorizer)
    if err != nil {
        panic("add error handling")
    }

    // Create an ingestion instance
    // Pass the client, the name of the database, and the name of table you wish to ingest into.
    in, err := ingest.New(client, dbName, tableName)
    if err != nil {
        panic("add error handling")
    }

    // Go currently only supports streaming from a byte array with a maximum size of 4 MB.
    jsonEncodedData := []byte("{\"a\":  1, \"b\":  10}\n{\"a\":  2, \"b\":  20}")

    // Ingestion from a stream commits blocks of fully formed data encodes (JSON, AVRO, ...) into Kusto:
    if err := in.Stream(context.Background(), jsonEncodedData, ingest.JSON, mappingName); err != nil {
        panic("add error handling")
    }
}

import com.microsoft.Azure Synapse Analytics.kusto.data.auth.ConnectionStringBuilder;
import com.microsoft.Azure Synapse Analytics.kusto.ingest.IngestClient;
import com.microsoft.Azure Synapse Analytics.kusto.ingest.IngestClientFactory;
import com.microsoft.Azure Synapse Analytics.kusto.ingest.IngestionProperties;
import com.microsoft.Azure Synapse Analytics.kusto.ingest.result.OperationStatus;
import com.microsoft.Azure Synapse Analytics.kusto.ingest.source.CompressionType;
import com.microsoft.Azure Synapse Analytics.kusto.ingest.source.StreamSourceInfo;
import java.io.FileInputStream;
import java.io.InputStream;

public class FileIngestion {
    public static void main(String[] args) throws Exception {
        String poolPath = "https://<Poolname>.<WorkspaceName>.kusto.windows.net";
        String appId = "<appId>";
        String appKey = "<appKey>";
        String appTenant = "<appTenant>";
        String dbName = "<dbName>";
        String tableName = "<tableName>";

        // Build connection string and initialize
        ConnectionStringBuilder csb =
            ConnectionStringBuilder.createWithAadApplicationCredentials(
                poolPath,
                appId,
                appKey,
                appTenant
            );

        // Initialize client and its properties
        IngestClient client = IngestClientFactory.createClient(csb);
        IngestionProperties ingestionProperties =
            new IngestionProperties(
                dbName,
                tableName
            );

        // Ingest from a compressed file
        // Create Source info
        InputStream zipInputStream = new FileInputStream("MyFile.gz");
        StreamSourceInfo zipStreamSourceInfo = new StreamSourceInfo(zipInputStream);
        // If the data is compressed
        zipStreamSourceInfo.setCompressionType(CompressionType.gz);
        // Ingest from stream
        OperationStatus status = client.ingestFromStream(zipStreamSourceInfo, ingestionProperties).getIngestionStatusCollection().get(0).status;
    }
}

1. В Synapse Studio в области слева выберите элемент Разработка.

2. В разделе Скрипты KQL выберите + (Добавить новый ресурс) >Скрипт KQL. В области справа можно присвоить имя вашему сценарию.

3. В меню Подключение к выберите contosodataexplorer.

4. В меню Использование базы данных выберите TestDatabase.

5. Вставьте следующую команду и выберите Выполнить, чтобы создать таблицу.

   .delete table TestTable policy streamingingestion
   

6. На портале Azure перейдите к пулу Data Explorer.

7. Под разделом Параметры выберите пункт Конфигурации.

8. В области Конфигурации выберите Вкл. , чтобы включить Прием потоковой передачи.

9. Щелкните Сохранить.

1. На портале Azure перейдите к пулу Data Explorer и выберите Запрос.

2. Чтобы удалить политику приема потоковой передачи из таблицы, скопируйте следующую команду в Панель запросов и выберите Выполнить.

   .delete table TestTable policy streamingingestion
   

3. Под разделом Параметры выберите пункт Конфигурации.

4. В области Конфигурации выберите Выкл. , чтобы включить Прием потоковой передачи.

5. Нажмите кнопку Сохранить.