Öğretici: Tablo güncelleştirme ilkelerini kullanarak verileri yönlendirme

  • Makale

Kaynak verileriniz basit ve hızlı dönüştürmeler içeriyorsa, bunları işlem hattında bir olay akışı kullanarak yukarı akış gerçekleştirmek en iyisidir. Ancak, bu yaklaşım karmaşık olan veya çalışması için özel işlevler gerektiren diğer dönüştürmelerde iyi çalışmayabilir.

Bu öğreticide şunların nasıl yapıldığını öğreneceksiniz:

Bu öğreticideki örnekte , veri alımı sırasında verileri zenginleştirmek, temizlemek ve dönüştürmek için karmaşık dönüştürmeler gerçekleştirmek üzere veri yönlendirme için güncelleştirme ilkelerinin nasıl kullanılacağı gösterilmektedir. Diğer yaygın kullanım örneklerinin listesi için bkz. Tablo güncelleştirme ilkeleri için yaygın kullanım örnekleri.

Önkoşullar

1 - Tablo oluşturma ve ilkeleri güncelleştirme

Aşağıdaki adımlar kaynak tablo, dönüştürme işlevleri, hedef tablolar ve güncelleştirme ilkeleri oluşturma konusunda size yol gösterir. Öğreticide, karmaşık dönüştürmeler yapmak ve sonuçları bir veya daha fazla hedef tabloya kaydetmek için tablo güncelleştirme ilkelerinin nasıl kullanılacağı gösterilmektedir. Örnekte Raw_Table adlı tek bir kaynak tablo ve Device_Telemetry, Device_Alarms ve Error_Log adlı üç hedef tablo kullanılır.

  1. Raw_Table adlı bir tablo oluşturmak için aşağıdaki komutu çalıştırın.

    .create table Raw_Table (RawData: dynamic)

    Alınan verilerin kaydedildiği kaynak tablodur. Tabloda dinamik türünde RawData adlı tek bir sütun vardır. Dinamik tür, herhangi bir şema olmadan ham verileri olduğu gibi depolamak için kullanılır. Daha fazla bilgi için bkz. .create table command.

  2. Get_Telemetry, Get_AlarmsveLog_Error işlevleri adlı bir işlev oluşturmak için aşağıdaki komutu çalıştırın.

    .execute database script <|
  .create-or-alter function Get_Telemetry() {
    Raw_Table
    | where todynamic(RawData).MessageType == 'Telemetry'
    | extend
      Timestamp = unixtime_seconds_todatetime(tolong(RawData.Timestamp)),
      DeviceId = tostring(RawData.DeviceId),
      DeviceType = tostring(RawData.DeviceType),
      SensorName = tostring(RawData.SensorName),
      SensorValue = toreal(RawData.SensorValue),
      SensorUnit = tostring(RawData.SensorUnit)
    | project-away RawData
  }
  .create-or-alter function Get_Alarms() {
    Raw_Table
    | where RawData.MessageType == 'Alarms'
    | extend
      Timestamp = unixtime_seconds_todatetime(tolong(RawData.Timestamp)),
      DeviceId = tostring(RawData.DeviceId),
      DeviceType = tostring(RawData.DeviceTpe) ,
      AlarmType = tostring(RawData.AlarmType)
    | project-away RawData
  }
  .create-or-alter function Log_Error() {
    Raw_Table
    | where RawData.MessageType !in ('Telemetry', 'Alarms')
    | extend
      TimeStamp = datetime(now),
      ErrorType = 'Unknown MessageType'
    | project TimeStamp, RawData, ErrorType
  }

    Güncelleştirme ilkesi oluştururken, yürütme için bir satır içi betik belirtebilirsiniz. Ancak, dönüştürme mantığını bir işlev olarak kapsüllemenizi öneririz. İşlev kullanmak kod bakımını geliştirir. Yeni veriler geldiğinde, verileri dönüştürmek için işlevi yürütülür. İşlev, birden çok güncelleştirme ilkesinde yeniden kullanılabilir. Daha fazla bilgi için bkz. .create function komutu.

  3. Hedef tabloları oluşturmak için aşağıdaki komutu çalıştırın.

    .execute database script <|
  .create table Device_Telemetry (Timestamp: datetime, DeviceId: string, DeviceType: string, SensorName: string, SensorValue: real, SensorUnit: string)
  .set-or-append Device_Alarms <| Get_Alarms | take 0
  .set-or-append Error_Log <| Log_Error | take 0

    Hedef tablo, dönüştürme işlevinin çıkışıyla aynı şemaya sahip olmalıdır. Hedef tabloları aşağıdaki yollarla oluşturabilirsiniz:

    • .create table komutunu kullanarak ve Device_Telemetry tablosunun oluşturulmasıyla gösterildiği gibi şemayı el ile belirtme. Ancak bu yaklaşım hataya açık ve zaman alıcı olabilir.
    • .set-or-append Verileri dönüştürmek için zaten bir işlev oluşturduysanız komutunu kullanın. Bu yöntem, işlevinin yalnızca şemayı döndürdüğünden emin olmak için kullanarak take 0 işlevinin çıkışıyla aynı şemaya sahip yeni bir tablo oluşturur. Daha fazla bilgi için bkz . .set-or-append komutu.

  4. Hedef tablolar için güncelleştirme ilkelerini oluşturmak için aşağıdaki komutu çalıştırın

    .execute database script <|
  .alter table Device_Telemetry policy update "[{\"IsEnabled\":true,\"Source\":\"Raw_Table\",\"Query\":\"Get_Telemetry\",\"IsTransactional\":false,\"PropagateIngestionProperties\":true,\"ManagedIdentity\":null}]"
  .alter table Device_Alarms policy update "[{\"IsEnabled\":true,\"Source\":\"Raw_Table\",\"Query\":\"Get_Alarms\",\"IsTransactional\":false,\"PropagateIngestionProperties\":true,\"ManagedIdentity\":null}]"
  .alter table Error_Log policy update "[{\"IsEnabled\":true,\"Source\":\"Raw_Table\",\"Query\":\"Log_Error\",\"IsTransactional\":false,\"PropagateIngestionProperties\":true,\"ManagedIdentity\":null}]"

    .alter table policy update Komut, kaynak tabloyu, dönüştürme işlevini ve hedef tabloyu bağlamak için kullanılır. Güncelleştirme ilkesi hedef tabloda oluşturulur ve kaynak tablo ile dönüştürme işlevini belirtir. Daha fazla bilgi için bkz. .alter table policy update komutu.

2 - Örnek verileri alma

Güncelleştirme ilkelerini test etmek için komutunu kullanarak örnek verileri kaynak tabloya .set-or-append alabilirsiniz. Daha fazla bilgi için bkz. Sorgudan veri alma.

.set-or-append Raw_Table <|
  let Raw_Stream = datatable(RawData: dynamic)
    [
    dynamic({"TimeStamp": 1691757932, "DeviceId": "Sensor01", "MessageType": "Telemetry", "DeviceType": "Laminator", "SensorName": "Temperature", "SensorValue": 78.3, "SensorUnit": "Celcius"}),
    dynamic({"TimeStamp": 1691757932, "DeviceId": "Sensor01", "MessageType": "Alarms", "DeviceType": "Laminator", "AlarmType": "Temperature threshold breached"}),
    dynamic({"TimeStamp": 1691757932, "DeviceId": "Sensor01", "MessageType": "Foo", "ErrorType": "Unknown"})
  ];
  Raw_Stream

3 - Sonuçları doğrulama

Sonuçları doğrulamak için bir sorgu çalıştırarak verilerin dönüştürüldüğünü ve hedef tablolara yönlendirildiğini doğrulayabilirsiniz. Aşağıdaki örnekte, union hedef tablolardaki kaynak ve sonuçları tek bir sonuç kümesinde birleştirmek için işleci kullanılır.

Raw_Table | summarize Rows=count() by TableName = "Raw_Table"
| union (Device_Telemetry | summarize Rows=count() by TableName = "Device_Telemetry")
| union (Device_Alarms | summarize Rows=count() by TableName = "Device_Alarms")
| union (Error_Log | summarize Rows=count() by TableName = "Error_Log")
| sort by Rows desc

Çıkış

Raw_Table üç satırın olduğu ve hedef tabloların her biri bir satıra sahip olduğu aşağıdaki çıkışı görmeniz gerekir.

TableName Satırlar
Raw_Table 3
Error_Log 1
Device_Alarms 1
Device_Telemetry 1

Kaynakları temizleme

Bu öğreticide oluşturulan tabloları ve işlevleri temizlemek için veritabanınızda aşağıdaki komutu çalıştırın.

.execute database script <|
  .drop table Raw_Table
  .drop table Device_Telemetry
  .drop table Device_Alarms
  .drop table Error_Log
  .drop function Get_Telemetry
  .drop function Get_Alarms
  .drop function Log_Error

İlgili içerik