Azure 事件中心数据连接

Azure 事件中心是大数据流式处理平台和事件引入服务。 Azure 数据资源管理器通过客户管理的事件中心提供持续引入。

事件中心引入管道通过几个步骤将事件传输到 Azure 数据资源管理器。 首先，在 Azure 门户中创建事件中心。 然后，创建 Azure 数据资源管理器目标表，使用给定的引入属性将特定格式的数据引入到该表中。 事件中心连接需要知道事件路由。 根据事件系统属性，可能会使用选定的属性嵌入数据。 与事件中心建立连接，以创建事件中心并发送事件。 可以通过 Azure 门户使用 C# 或 Python 以编程方式管理此过程，也可以使用 Azure 资源管理器模板来这样做。

有关 Azure 数据资源管理器中数据引入的常规信息，请参阅 Azure 数据资源管理器数据引入概述。

Azure 数据资源管理器数据连接身份验证机制

• 基于托管标识的数据连接（推荐）：基于托管标识的数据连接是连接到数据源的最安全方式。 它完全控制从数据源提取数据的能力。 使用托管标识设置数据连接需要执行以下步骤：

  1. 将一个托管标识添加到群集。
  2. 向该托管标识授予对数据源的权限。 若要从 Azure 事件中心提取数据，托管标识必须具有 Azure 事件中心数据接收器权限。
  3. 在目标数据库上设置托管标识策略。
  4. 创建使用托管标识身份验证的数据连接来提取数据。

  注意

  如果从数据源中删除托管标识权限，则数据连接将不再有效，并且无法从数据源中提取数据。

• 基于密钥的数据连接：如果未为数据连接指定托管标识身份验证，则连接会自动默认为基于密钥的身份验证。 基于密钥的连接使用资源连接字符串（如 Azure事件中心连接字符串）提取数据。 Azure 数据资源管理器获取指定资源的资源连接字符串，并安全地保存它。 该连接字符串之后用于从数据源提取数据。

  注意

  如果轮换密钥，则数据连接将不再有效，并且无法从数据源中提取数据。 若要解决此问题，请更新或重新创建数据连接。

数据格式

• 将以 EventData 对象的形式从事件中心读取数据。

• 请参阅支持的格式。

  注意

  • 从事件中心引入不支持 RAW 格式。
  • 不支持 Azure 事件中心架构注册表和无架构 Avro。

• 可使用 GZip 压缩算法来压缩数据。 可使用引入属性动态指定 Compression，或者在静态数据连接设置中指定。

  注意

  压缩格式（Avro、Parquet、ORC、ApacheAvro 和 W3CLOGFILE）不支持数据压缩。 压缩数据不支持自定义编码和嵌入式系统属性。

事件中心属性

Azure 数据资源管理器支持以下事件中心属性：

• 一组封闭的引入属性，它们可以帮助将事件路由到相关的表。
• 一组封闭的事件系统属性，可以根据给定的映射将它们嵌入在数据中。

引入属性

引入属性会指示引入过程、数据路由到的位置以及数据处理方式。 可以使用 EventData.Properties 指定事件引入的引入属性。 可以设置以下属性：

事件路由

在与群集建立数据连接时，可以指定要将引入的数据发送到何处的路由。 默认路由是发送到与目标数据库关联的连接字符串中指定的目标表。 数据的默认路由也称为静态路由。 可以通过设置上述事件数据属性来为数据指定替代路由。

将事件数据路由到备用数据库

默认已禁用将数据路由到备用数据库。 若要将数据发送到其他数据库，必须先将连接设置为多数据库连接。 可以在 Azure 门户、C#、Python 或 ARM 模板中执行此操作。 用于允许数据库路由的用户、组、服务主体或托管标识在群集上必须至少具有参与者角色和写入权限。

若要指定备用数据库，请设置数据库引入属性。

将事件数据路由到备用表

若要为每个事件指定备用表，请设置“表”、“格式”、“压缩”和映射引入属性。 连接将按照 EventData.Properties 中指定的要求动态路由引入的数据，并重写此事件的静态属性。

以下示例演示如何设置事件中心详细信息，并将天气指标数据发送到备用数据库 (MetricsDB) 和表 (WeatherMetrics)。 数据采用 JSON 格式，mapping1 是在表 WeatherMetrics 中预定义的。

// This sample uses Azure.Messaging.EventHubs which is a .Net Framework library.
await using var producerClient = new EventHubProducerClient("<eventHubConnectionString>");
// Create the event and add optional "dynamic routing" properties
var eventData = new EventData(Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(
    new { Timestamp = DateTime.UtcNow, MetricName = "Temperature", Value = 32 }
)));
eventData.Properties.Add("Database", "MetricsDB");
eventData.Properties.Add("Table", "WeatherMetrics");
eventData.Properties.Add("Format", "json");
eventData.Properties.Add("IngestionMappingReference", "mapping1");
eventData.Properties.Add("Tags", "['myDataTag']");
var events = new[] { eventData };
// Send events
await producerClient.SendAsync(events);

事件系统属性映射

系统属性在事件排队时存储由事件中心服务设置的属性。 与事件中心建立的数据连接可以基于给定的映射将选定的一组系统属性嵌入到在表中引入的数据。

事件中心服务公开以下系统属性：

使用 IoT Central 事件中心时，还可以在有效负载中嵌入 IoT 中心系统属性。 有关完整列表，请参阅 IoT 中心系统属性。

如果在表的“数据源”部分中选择了“事件系统属性”，则必须在表架构和映射中包含这些属性。

架构映射示例

表架构映射示例

如果数据包含三列（Timespan、Metric 和 Value）并且包含的属性是 x-opt-enqueued-time 和 x-opt-offset，请使用以下命令创建或更改表架构：

    .create-merge table TestTable (TimeStamp: datetime, Metric: string, Value: int, EventHubEnqueuedTime:datetime, EventHubOffset:string)

CSV 映射示例

运行以下命令，将数据添加到记录的开头。 记下序号值。

    .create table TestTable ingestion csv mapping "CsvMapping1"
    '['
    '   { "column" : "Timespan", "Properties":{"Ordinal":"2"}},'
    '   { "column" : "Metric", "Properties":{"Ordinal":"3"}},'
    '   { "column" : "Value", "Properties":{"Ordinal":"4"}},'
    '   { "column" : "EventHubEnqueuedTime", "Properties":{"Ordinal":"0"}},'
    '   { "column" : "EventHubOffset", "Properties":{"Ordinal":"1"}}'
    ']'

JSON 映射示例

使用系统属性映射添加数据。 运行以下命令：

    .create table TestTable ingestion json mapping "JsonMapping1"
    '['
    '    { "column" : "Timespan", "Properties":{"Path":"$.timestamp"}},'
    '    { "column" : "Metric", "Properties":{"Path":"$.metric"}},'
    '    { "column" : "Value", "Properties":{"Path":"$.value"}},'
    '    { "column" : "EventHubEnqueuedTime", "Properties":{"Path":"$.x-opt-enqueued-time"}},'
    '    { "column" : "EventHubOffset", "Properties":{"Path":"$.x-opt-offset"}}'
    ']'

事件中心捕获 Avro 文件的架构映射

使用事件中心数据的一种方法是通过 Azure Blob 存储或 Azure Data Lake Storage 中的 Azure 事件中心捕获事件。 然后，可以在使用 Azure 数据资源管理器中的事件网格数据连接写入捕获文件时引入这些文件。

捕获文件的架构与发送到事件中心的原始事件的架构不同。 在设计目标表架构时，应考虑到这种差异。 具体而言，事件负载在捕获文件中表示为字节数组，并且事件网格 Azure 数据资源管理器数据连接不会自动解码此数组。 有关事件中心 Avro 捕获数据的文件架构的详细信息，请参阅探索 Azure 事件中心中捕获的 Avro 文件。

若要正确解码事件负载，请执行以下操作：

1. 将捕获事件的 Body 字段映射到目标表中 dynamic 类型的列。
2. 应用一项更新策略，该策略使用 unicode_codepoints_to_string() 函数将字节数组转换为可读字符串。

引入自定义属性

从事件中心引入事件时，数据将取自事件数据对象的 body 节。 但是，事件中心自定义属性是在该对象的 properties 节中定义的，不会引入。 若要引入客户属性，必须将其嵌入到该对象的 body 节中的数据。

以下示例将包含事件中心定义的自定义属性 customProperty 的事件数据对象（左）与引入时所需的嵌入属性（右）进行比较。

{
"body":{
"value": 42
},
"properties":{
"customProperty": "123456789"
}
}

{
"body":{
"value": 42,
"customProperty": "123456789"
}
}

可使用以下方法之一将自定义属性嵌入到事件数据对象的 body 节中的数据：

• 在事件中心内创建事件数据对象时，将自定义属性嵌入为该对象的 body 节中的数据的一部分。
• 使用 Azure 流分析来处理来自事件中心的事件并将自定义属性嵌入到事件数据中。 在 Azure 流分析中，可以使用 Azure 数据资源管理器输出连接器本机引入数据，或者将数据路由到另一个事件中心并从中路由到群集。
• 使用 Azure Functions 添加自定义属性，然后引入数据。

跨区域事件中心数据连接

为获得最佳性能，请在群集所在的同一区域中创建以下所有资源。 如果没有其他替代方法，请考虑使用高级或专用事件中心层级。 可以在此处找到事件中心层级对比。

创建事件中心

创建事件中心（如果还没有事件中心）。 可以通过 Azure 门户、使用 C# 或 Python 以编程方式或使用 Azure 资源管理器模板来管理到数据中心的连接。

发送事件

请参阅可生成数据并将其发送到事件中心的示例应用。

设置异地灾难恢复解决方案

事件中心提供异地灾难恢复解决方案。 Azure 数据资源管理器不支持 Alias 事件中心命名空间。 若要在解决方案中实现异地灾难恢复，请创建两个事件中心数据连接：一个用于主命名空间，另一个用于辅助命名空间。 Azure 数据资源管理器将侦听这两个事件中心连接。

相关内容

• 将数据从事件中心引入 Azure 数据资源管理器
• 使用 C# 为 Azure 数据资源管理器创建事件中心数据连接
• 使用 Python 为 Azure 数据资源管理器创建事件中心数据连接
• 使用 Azure 资源管理器模板为 Azure 数据资源管理器创建事件中心数据连接
• 使用 Azure 数据资源管理器引入和查询 Azure Monitor 日志