Azure Synapse Data Explorer プールでストリーミング インジェストを構成する (プレビュー)

「Synapse Studio を使用してデータ エクスプローラー プールを作成する」の手順を使用してデータ エクスプローラー プールを作成するときに、[追加の設定] タブで [ストリーミング インジェスト>Enabled] を選択します。

「Azure portal を使用してデータ エクスプローラー プールを作成する」の手順を 使用してデータ エクスプローラー プールを作成するときに、[ 追加の設定 ] タブで [ ストリーミング インジェスト>Enabled] を選択します。

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 ポータルで、データ エクスプローラー プールに移動します。

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 で、データ エクスプローラー プールに移動します。

7. [設定] で、[構成] を選択します。

8. [構成] ウィンドウで、[オン] を選択してストリーミング インジェストを有効にします。

9. 保存 を選択します。

1. Azure portal で、データ エクスプローラー プールに移動し、[クエリ] を選択 します。

2. テーブルからストリーミング インジェスト ポリシーを削除するには、次のコマンドを クエリ ペイン にコピーし、[ 実行] を選択します。

   .delete table TestTable policy streamingingestion
   

3. [設定] で、[構成] を選択します。

4. [構成] ウィンドウで、[オフ] を選択してストリーミング インジェストを無効にします。

5. 保存 を選択します。