Azure Synapse Data Explorer 풀에서 스트리밍 수집 구성(미리 보기)

Synapse Studio를 사용하여 Data Explorer 풀 만들기의 단계를 통해 Data Explorer 풀을 만드는 동안 추가 설정 탭에서 스트리밍 수집>사용을 선택합니다.

Azure Portal을 사용하여 Data Explorer 풀 만들기의 단계를 통해 Data Explorer 풀을 만드는 동안 추가 설정 탭에서 스트리밍 수집>사용을 선택합니다.

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 Portal에서 Data Explorer 풀로 이동합니다.

7. 설정에서 구성을 선택합니다.

8. 구성 창에서 켜기를 선택하여 스트리밍 수집을 사용하도록 설정합니다.

9. 저장을 선택합니다.

1. Azure Portal에서 Data Explorer 풀로 이동하여 쿼리를 선택합니다.

2. 테이블에서 스트리밍 수집 정책을 삭제하려면 다음 명령을 쿼리 창에 복사하고 실행을 선택합니다.

   .delete table TestTable policy streamingingestion
   

3. 설정에서 구성을 선택합니다.

4. 구성 창에서 끄기를 선택하여 스트리밍 수집을 사용하지 않도록 설정합니다.

5. 저장을 선택합니다.