在 Azure Synapse 数据资源管理器池（预览版）上配置流式引入

重要

Azure Synapse Analytics 数据资源管理器（预览版）将于 2025 年 10 月 7 日停用。在此日期之后，Synapse 数据资源管理器上运行的工作负荷将被删除，关联的应用程序数据将丢失。强烈建议迁移到 Microsoft Fabric 中的 Eventhouse 。

Microsoft云迁移工厂（CMF）计划旨在帮助客户迁移到 Fabric。该计划向客户免费提供动手键盘资源。这些资源在 6-8 周内分配，并具有预定义和同意的范围。客户提名可以通过 Microsoft 帐户团队接受，或者直接提交《帮助请求》给 CMF 团队。

引入和查询之间需要低延迟时，流式引入可用于加载数据。请考虑在以下方案中使用流式引入：

需要小于一秒的延迟。
需要优化多个表的操作处理，其中进入每个表的数据流相对较小（每秒几条记录），但总体数据引入量较大（每秒成千上万条记录）。

如果每个表中的数据流较高（每小时超过 4 GB），请考虑使用批处理引入。

若要详细了解不同的引入方法，请参阅数据引入概述。

选择适当的流式引入类型

支持两种流式引入类型：

引入类型	Description
事件中心或 IoT 中心	中心配置为表流数据源。有关设置这些设置的信息，请参阅事件中心。
自定义导入	自定义引入要求编写使用其中一个 Azure Synapse 数据资源管理器客户端库的应用程序。使用本主题中的信息配置自定义数据导入。还可以发现 C# 流式引入示例应用程序很有用。

使用下表可帮助你选择适合你的环境的引入类型：

标准	事件中心/IoT 中心	自定义引入
引入启动与可用于查询的数据之间的数据延迟	延迟更长	更短的延迟
开发成本	快速而简单的设置，无需开发开销	建立应用程序以引入数据、确保数据一致性和处理错误需要高昂的开发开销

注释

如果 Synapse 工作区在启用了数据外泄保护的情况下适用托管虚拟网络，则将数据从事件中心引入数据资源管理器池将不起作用。

先决条件

一份 Azure 订阅。创建免费 Azure 帐户。
使用 Synapse Studio 或 Azure 门户创建数据资源管理器池

创建“Data Explorer”数据库。

在 Synapse Studio 的左窗格中，选择“数据”。

选择+（添加新资源）>数据探索池，并使用以下信息：

设置	建议值	Description
池名称	contosodataexplorer	要使用的数据资源管理器池的名称
Name	TestDatabase	该数据库名称在群集中必须是唯一的。
默认保留期	365	保证数据可供查询的时间跨度（以天为单位）。时间跨度从引入数据时算起。
默认缓存期	31	将频繁查询的数据保存在 SSD 存储或 RAM 中（而不是长期存储）的时间跨度（以天为单位）。

选择“创建”以创建数据库。创建过程通常需要不到一分钟的时间。

获取查询和数据引入终结点。
1. 在 Synapse Studio 的左侧窗格中，选择管理>数据资源管理器池。
2. 选择要用于查看其详细信息的数据资源管理器池。
3. 记下查询和数据引入终结点。在配置到数据资源管理器池的连接时，请使用查询终结点作为群集。为数据引入配置 SDK 时，请使用数据引入终结点。

性能和操作注意事项

可能会影响流式引入的主要因素包括：

计算规范：流式引入性能和容量随数据资源管理器池大小的增加而扩展。并发引入请求数限制为每个核心 6 个。例如，对于 16 个核心工作负荷类型（例如计算优化（大型）和存储优化（大型），最大支持的负载为 96 个并发引入请求。对于两种核心工作负荷类型，例如计算优化（特小），支持的最大负载是 12 个并发引入请求。
数据大小限制：流式引入请求的数据大小限制为 4 MB。
架构更新：流式引入服务可能需要长达五分钟才能进行架构更新，例如创建和修改表和引入映射。有关详细信息，请参阅流式引入和架构更改。
SSD 容量：在数据资源管理器池上启用流式引入，即使数据未通过流式引入，也会使用数据资源管理器池计算机的本地 SSD 磁盘的一部分进行流式引入数据，并减少可用于热缓存的存储。

在数据资源管理器池上启用流式引入

在使用流式引入之前，必须在数据资源管理器池上启用该功能并定义流式引入策略。可以在创建数据资源管理器池时启用该功能，或将其添加到现有的数据资源管理器池。

警告

在启用流式引入之前查看限制。

创建新的数据资源管理器池时启用流式引入

使用 Azure Synapse Studio 或 Azure 门户创建新的数据资源管理器池时，可以启用流式引入。

演播室
Portal

使用 Synapse Studio 创建数据资源管理器池中的步骤创建数据资源管理器池时，在“其他设置”选项卡中，选择“已启用>”。

在 Azure Synapse 数据资源管理器中创建数据资源管理器池时启用流式引入。

在现有数据探索者池上启用流式引入

如果有一个现成的数据资源管理器池，可以使用 Azure 门户启用流式引入。

在 Azure 门户中，转到数据资源管理器池。
在 “设置”中，选择“ 配置”。
在 “配置 ”窗格中，选择 “打开 ”以启用 流式引入。
选择“保存”。

创建目标表并定义策略

创建一个表来接收流式引入数据，并使用 Azure Synapse Studio 或 Azure 门户定义其相关策略。

演播室
Portal

在 Synapse Studio 的左侧窗格中，选择“开发”。
在KQL 脚本下，选择+（添加新资源）>KQL 脚本。可在右侧窗格中命名脚本。
在“连接到”菜单中，选择“contosodataexplorer”。
在“使用数据库”菜单中，选择“TestDatabase”。

粘贴以下命令，然后选择“运行”以创建表。

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

将以下命令之一复制到 “查询”窗格中 ，然后选择“ 运行”。这会定义对创建的表或包含表的数据库的流式引入策略。

小窍门

在数据库级别定义的策略适用于数据库中的所有现有表和将来表。
- 若要在创建的表上定义策略，请使用：
```
.alter table TestTable policy streamingingestion enable
```
- 若要在包含所创建表的数据库上定义策略，请使用：
```
.alter database StreamingTestDb policy streamingingestion enable
```

在 Azure Synapse Analytics 门户中，导航到数据资源管理器池。
选择查询。
若要创建将通过流式引入接收数据的表，请将以下命令复制到 “查询”窗格中 ，然后选择“ 运行”。
```
.create table TestTable (TimeStamp: datetime, Name: string, Metric: int, Source:string)
```
将以下命令之一复制到 “查询”窗格中 ，然后选择“ 运行”。这会定义对创建的表或包含表的数据库的流式引入策略。

小窍门

在数据库级别定义的策略适用于数据库中的所有现有表和将来表。
- 若要在创建的表上定义策略，请使用：
```
.alter table TestTable policy streamingingestion enable
```
- 若要在包含所创建表的数据库上定义策略，请使用：
```
.alter database StreamingTestDb policy streamingingestion enable
```

创建流式引入应用程序，以将数据引入数据资源管理器池

使用首选语言创建用于将数据引入到数据资源管理器池的应用程序。对于 poolPath 变量，请使用在先决条件中记下的查询终结点。

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

在数据资源管理器池上禁用流式引入

警告

禁用流式引入可能需要几个小时。

在数据资源管理器池上禁用流式引入之前，请从所有相关表和数据库中删除流式引入策略。删除流式引入策略会触发数据资源管理器池中的数据重新排列。流式引入数据将从初始存储移到列存储中的永久存储（盘区或分片）。此过程可能需要几秒钟到几个小时，具体取决于初始存储中的数据量。

删除流式引入策略

可以使用 Azure Synapse Studio 或 Azure 门户删除流式引入策略。

演播室
Portal

在 Synapse Studio 的左侧窗格中，选择“开发”。
在KQL 脚本下，选择+（添加新资源）>KQL 脚本。可在右侧窗格中命名脚本。
在“连接到”菜单中，选择“contosodataexplorer”。
在“使用数据库”菜单中，选择“TestDatabase”。
粘贴以下命令，然后选择“运行”以创建表。
```
.delete table TestTable policy streamingingestion
```
在 Azure 门户中，转到数据资源管理器池。
在 “设置”中，选择“ 配置”。
在 “配置 ”窗格中，选择 “打开 ”以启用 流式引入。
选择“保存”。

在 Azure 门户中，转到数据资源管理器池并选择 “查询”。
若要从表中删除流式引入策略，请将以下命令复制到 “查询”窗格中 ，然后选择“ 运行”。
```
.delete table TestTable policy streamingingestion
```
在 “设置”中，选择“ 配置”。
在 “配置 ”窗格中，选择“ 关闭 ”以禁用 流式引入。
选择“保存”。

局限性

如果数据库本身或其任何表已定义并启用流式引入策略，则数据库不支持数据库游标。
必须预先创建数据映射，以便在流式引入中使用。单个流式引入请求不能容纳内联数据映射。
盘区标记不能在流式引入数据上设置。
更新策略。更新策略只能引用源表中新引入的数据，而不能引用数据库中的任何其他数据或表。
如果对数据库的任何表使用流式引入，则不能将此数据库用作关注者数据库的领导者，也不能用作 Azure Synapse Analytics Data Share 的数据提供程序。

后续步骤

Last updated on 2025-03-24

通过