Formato dei messaggi JSON - Modificare lo streaming di eventi

Si applica a: SQL Server 2025 (17.x) Database SQL di AzureIstanza gestita di SQL di AzureDatabase SQL in Microsoft Fabric

Questo articolo descrive il formato di messaggi CloudEvents che viene trasmesso su Hub eventi di Azure o Fabric Eventstream quando si utilizza la funzione change event streaming (CES) in SQL Server 2025 (17.x), database SQL di Azure, Istanza gestita di SQL di Azure, e database SQL in Microsoft Fabric.

Informazioni generali

Lo streaming di eventi di modifica emette eventi che seguono la specifica CloudEvents , così puoi facilmente integrarli con sistemi basati su eventi. Tutti i cloudEvent CES contengono 11 attributi (campi). Puoi configurare CES per serializzare l'intero CloudEvent, incluso l'attributo data , come binario JSON nativo o Avro. Gli eventi JSON nativi non contengono sezioni binarie Avro. In entrambi i formati di serializzazione, l'attributo data ha un tipo byte-array. I byte utilizzano codifica binaria JSON o Avro secondo il formato di serializzazione selezionato e seguono lo schema Avro dell'attributo dati CES.

Specifiche e risorse correlate

Quando applicabile, le descrizioni in questa sezione provengono dalla specifica CloudEvent, che include maggiori dettagli.

Attributi

• specversion:

  • Tipo di dati: String
  • Attributo CloudEvent obbligatorio
  • Versione della specifica CloudEvents usata dall'evento. Questa versione consente l'interpretazione del contesto.

• type

  • Tipo di dati: String
  • Attributo CloudEvent obbligatorio
  • Contiene un valore che descrive il tipo di evento correlato all'occorrenza di origine. Il formato di questo valore è definito dal produttore e può includere informazioni come la versione del tipo. Per maggiori informazioni, vedi Versioning of CloudEvents.
  • Per cambiare eventi di streaming eventi, il tipo attuale è: com.microsoft.SQL.CES.DML.V{n}, dove {n} indica la versione dello schema di eventi DML di cambio evento di streaming eventi di Microsoft.
    • L'ultima versione attuale dello schema è 1.

• source

  • Tipo di dati: String
  • Attributo CloudEvent obbligatorio
  • Identifica il contesto in cui si è verificato un evento. La combinazione di sorgente e ID deve essere unica per ogni evento. Attualmente, questo campo viene sempre inviato come \/ eventi streamati da SQL.

• id

  • Tipo di dati: String
  • Attributo CloudEvent obbligatorio
  • Identifica l'evento. I produttori devono assicurarsi che la combinazione di origine e ID sia unica per ogni evento distinto. Se un evento duplicato viene reinviato ,ad esempio a causa di un errore di rete, potrebbe avere lo stesso ID. I consumer possono presupporre che gli eventi con origine e ID identici siano duplicati.

• logicalid

  • Tipo di dati: String
  • Attributo di estensione
  • Gli ID logici condivisi identificano i messaggi divisi (a causa delle restrizioni sulla dimensione dei messaggi degli Event Hubs).

• time

  • Tipo di dati: Timestamp
  • Attributo CloudEvent facoltativo
  • Timestamp UTC di quando il commit è avvenuto all'interno di una transazione SQL che originariamente attiva un evento streaming.

• datacontenttype

  • Tipo di dati: String
  • Attributo CloudEvent facoltativo
  • Tipo di contenuto del valore di dati. Questo attributo consente ai dati di trasportare qualsiasi tipo di contenuto, in base al quale il formato e la codifica possono differire da quello del formato di evento scelto. Ad esempio, un evento sottoposto a rendering usando il formato busta JSON potrebbe portare un payload XML nei dati e il consumer viene informato da questo attributo impostato su "application/xml". Le regole su come il contenuto dei dati viene reso per valori diversi datacontenttype sono definite nelle specifiche del formato degli eventi.

• operation

  • Tipo di dati: String
  • Attributo di estensione
  • Rappresenta il tipo di operazione SQL che è avvenuta:
    • INS per inserti
    • Aggiornamento per aggiornamenti
    • DEL per le eliminazioni

• segmentindex

  • Tipo di dati: Integer
  • Attributo di estensione
  • Indice di segmento, che indica la posizione del messaggio all'interno dei blocchi logici del messaggio. L'indice del segmento fornisce informazioni sulla posizione del messaggio nella sequenza di frammenti di messaggi logici. Questo campo è sempre presente. Usa logicalid, , e segmentindex campi per ordinare gli eventi in arrivo che rappresentano una grande suddivisione del payload SQL secondo il valore configurato finalsegmentmax_message_size_kb.

• finalsegment

  • Tipo di dati: Boolean
  • Attributo di estensione
  • Indica se questo segmento è l'ultimo segmento della sequenza. Questo campo è sempre presente e aiuta a identificare se un evento SQL è stato suddiviso in sottoeventi secondo il valore configurato max_message_size_kb .

• data

  • Tipo di dato: Array di byte
  • Attributo CloudEvent facoltativo
  • Contiene i dati specifici dell'evento del dominio che descrivono la modifica. Deserializzare i byte come binario JSON o Avro secondo il formato di serializzazione selezionato. I dati deserializzati seguono lo schema Avro dell'attributo dati CES. Per informazioni sui suoi campi, vedi Data attribute format.

Esempi

Esempio di messaggio JSON - Insert

{
  "specversion": "1.0",
  "type": "com.microsoft.SQL.CES.DML.V1",
  "source": "\/",
  "id": "56cb8ff3-5c55-4f3b-a7f7-b044d1933ef6",
  "logicalid": "1bf2756a-c15f-4d2e-a2d5-7d3f9dbf85b0:000000B1000008A80007:00000000000000000001",
  "time": "2026-08-07T16:25:00.890Z",
  "datacontenttype": "application\/json",
  "operation": "INS",
  "segmentindex": 0,
  "finalsegment": true,
  "data": "{\"eventsource\":{\"db\":\"EmployeesDb\",\"schema\":\"dbo\",\"tbl\":\"Employees\",\"cols\":[{\"name\":\"Id\",\"type\":\"int\",\"index\":0},{\"name\":\"FirstName\",\"type\":\"nvarchar(50)\",\"index\":1},{\"name\":\"LastName\",\"type\":\"nvarchar(50)\",\"index\":2},{\"name\":\"SignupDate\",\"type\":\"datetime2(7)\",\"index\":3}],\"pkkey\":[{\"columnname\":\"Id\",\"value\":\"8\"}],\"transaction\":{\"commitlsn\":\"000000B1:000008A8:0007\",\"beginlsn\":\"000000B1:000008A8:0003\",\"sequencenumber\":1,\"finalevent\":false,\"committime\":\"2026-08-07T16:25:00.890Z\"}},\"eventrow\":{\"old\":\"{}\",\"current\":\"{\\\"Id\\\":\\\"8\\\",\\\"FirstName\\\":\\\"Nikola\\\",\\\"LastName\\\":\\\"Nikolic\\\",\\\"SignupDate\\\":\\\"2026-08-07 16:25:00.8833333\\\"}\"}}"
}

Esempio di messaggio JSON - aggiornamento

{
  "specversion": "1.0",
  "type": "com.microsoft.SQL.CES.DML.V1",
  "source": "\/",
  "id": "19221db1-a1b5-4ec7-8937-3fdf9d762abb",
  "logicalid": "1bf2756a-c15f-4d2e-a2d5-7d3f9dbf85b0:000000B1000009300009:00000000000000000001",
  "time": "2026-08-07T16:30:10.123Z",
  "datacontenttype": "application\/json",
  "operation": "UPD",
  "segmentindex": 0,
  "finalsegment": true,
  "data": "{\"eventsource\":{\"db\":\"EmployeesDb\",\"schema\":\"dbo\",\"tbl\":\"Employees\",\"cols\":[{\"name\":\"Id\",\"type\":\"int\",\"index\":0},{\"name\":\"FirstName\",\"type\":\"nvarchar(50)\",\"index\":1},{\"name\":\"LastName\",\"type\":\"nvarchar(50)\",\"index\":2},{\"name\":\"SignupDate\",\"type\":\"datetime2(7)\",\"index\":3}],\"pkkey\":[{\"columnname\":\"Id\",\"value\":\"8\"}],\"transaction\":{\"commitlsn\":\"000000B1:00000930:0009\",\"beginlsn\":\"000000B1:00000930:0002\",\"sequencenumber\":1,\"finalevent\":false,\"committime\":\"2026-08-07T16:30:10.123Z\"}},\"eventrow\":{\"old\":\"{\\\"Id\\\":\\\"8\\\",\\\"FirstName\\\":\\\"Nikola\\\",\\\"LastName\\\":\\\"Nikolic\\\",\\\"SignupDate\\\":\\\"2026-08-07 16:25:00.8833333\\\"}\",\"current\":\"{\\\"Id\\\":\\\"8\\\",\\\"FirstName\\\":\\\"Nikola\\\",\\\"LastName\\\":\\\"Nikolic-Smith\\\",\\\"SignupDate\\\":\\\"2026-08-07 16:25:00.8833333\\\"}\"}}"
}

Esempio di messaggio JSON - delete

{
  "specversion": "1.0",
  "type": "com.microsoft.SQL.CES.DML.V1",
  "source": "\/",
  "id": "520f9a65-43d7-47f2-94f5-7ea14df635ed",
  "logicalid": "1bf2756a-c15f-4d2e-a2d5-7d3f9dbf85b0:000000B1000009700008:00000000000000000001",
  "time": "2026-08-07T16:35:42.450Z",
  "datacontenttype": "application\/json",
  "operation": "DEL",
  "segmentindex": 0,
  "finalsegment": true,
  "data": "{\"eventsource\":{\"db\":\"EmployeesDb\",\"schema\":\"dbo\",\"tbl\":\"Employees\",\"cols\":[{\"name\":\"Id\",\"type\":\"int\",\"index\":0},{\"name\":\"FirstName\",\"type\":\"nvarchar(50)\",\"index\":1},{\"name\":\"LastName\",\"type\":\"nvarchar(50)\",\"index\":2},{\"name\":\"SignupDate\",\"type\":\"datetime2(7)\",\"index\":3}],\"pkkey\":[{\"columnname\":\"Id\",\"value\":\"8\"}],\"transaction\":{\"commitlsn\":\"000000B1:00000970:0008\",\"beginlsn\":\"000000B1:00000970:0003\",\"sequencenumber\":1,\"finalevent\":false,\"committime\":\"2026-08-07T16:35:42.450Z\"}},\"eventrow\":{\"old\":\"{\\\"Id\\\":\\\"8\\\",\\\"FirstName\\\":\\\"Nikola\\\",\\\"LastName\\\":\\\"Nikolic-Smith\\\",\\\"SignupDate\\\":\\\"2026-08-07 16:25:00.8833333\\\"}\",\"current\":\"{}\"}}"
}

Formato attributo dati

L'attributo data è un array di byte. Deserializzare i byte come binario JSON o Avro secondo il formato di serializzazione selezionato. In entrambi i formati, il record risultante Data segue lo schema Avro dell'attributo dati CES e contiene due attributi:

• eventsource
• eventrow

{
  "data": "{\"eventsource\": {}, \"eventrow\": {\"old\": \"{}\", \"current\": \"{}\"}}"
}

Le sezioni seguenti spiegano in modo più dettagliato gli attributi deserializzati.

eventsource

Descrive i metadati relativi al database e alla tabella in cui si è verificato l'evento:

• db

  • Tipo di dati: String
  • Descrizione: nome del database in cui si trova la tabella.
  • Esempio: EmployeesDb

• schema

  • Tipo di dati: String
  • Descrizione: schema del database che contiene la tabella.
  • Esempio: dbo

• tbl

  • Tipo di dati: String
  • Descrizione: tabella in cui si è verificato l'evento.
  • Esempio: Employees

• cols

  • Tipo di dati: Matrice
  • Descrizione: matrice che descrive in dettaglio le colonne della tabella.
    • name (filo): Il nome della colonna.
    • type (stringa): Il tipo di dato SQL della colonna, inclusa la sua lunghezza, precisione o scala quando applicabile. Gli esempi includono int, nvarchar(50)e datetime2(7).
    • index (intero): L'indice o la posizione della colonna nella tabella.

• pkkey

  • Tipo di dati: Matrice
  • Descrizione: rappresenta le colonne chiave primaria e i relativi valori per identificare la riga specifica.
    • columnname (stringa): Il nome della colonna usata nella chiave primaria.
    • value (stringa): Il valore della colonna usata nella chiave primaria. Questo valore aiuta a identificare in modo unico la riga.

• transaction

  • Tipo di dato: Oggetto
  • Descrizione: Descrive la transazione SQL che contiene l'operazione dei dati.
    • commitlsn (stringa): Il numero di sequenza log di commit (LSN) della transazione.
    • beginlsn (stringa): L'LSN iniziale della transazione.
    • sequencenumber (intero): Il numero sequenziale dell'operazione dati all'interno della transazione. Usa questo valore per ordinare gli eventi all'interno di una transazione.
    • finalevent (booleano): Non in uso. Questo campo ha sempre un valore di false.
    • committime (stringa): La data e l'ora in cui la transazione è stata commessa nel database.

eventrow

Descrive le modifiche a livello di riga e confronta i valori precedenti e correnti dei campi nel record.

• old (oggetto di cui è stato eseguito il wrapping in stringa): rappresenta i valori nella riga prima dell'evento.
  • Ogni coppia chiave-valore è costituita da:
    • <column_name>: (stringa): nome della colonna.
    • <column_value>: (string/int/etc.): valore precedente per la colonna.
• current (oggetto di cui è stato eseguito il wrapping in stringa): rappresenta i valori aggiornati nella riga dopo l'evento.
  • Analogamente all'oggetto precedente, con ogni coppia chiave-valore strutturata come:
    • <column_name> (string): nome della colonna.
    • <column_value> (string/int/etc.): valore nuovo o corrente per tale colonna.

CES CloudEvent schema Avro

{
  "type": "record",
  "name": "ChangeEvent",
  "fields": [
    {
      "name": "specversion",
      "type": "string"
    },
    {
      "name": "type",
      "type": "string"
    },
    {
      "name": "source",
      "type": "string"
    },
    {
      "name": "id",
      "type": "string"
    },
    {
      "name": "logicalid",
      "type": "string"
    },
    {
      "name": "time",
      "type": "string"
    },
    {
      "name": "datacontenttype",
      "type": "string"
    },
    {
      "name": "operation",
      "type": "string"
    },
    {
      "name": "segmentindex",
      "type": "int"
    },
    {
      "name": "finalsegment",
      "type": "boolean"
    },
    {
      "name": "data",
      "type": "bytes"
    }
  ]
}

CES data attribute schema Avro

Usa il seguente schema quando si deserializza l'array data di byte in CloudEvents binari JSON nativi e Avro:

{
  "name": "Data",
  "type": "record",
  "fields": [
    {
      "name": "eventsource",
      "type": {
        "name": "EventSource",
        "type": "record",
        "fields": [
          {
            "name": "db",
            "type": "string"
          },
          {
            "name": "schema",
            "type": "string"
          },
          {
            "name": "tbl",
            "type": "string"
          },
          {
            "name": "cols",
            "type": {
              "type": "array",
              "items": {
                "name": "Column",
                "type": "record",
                "fields": [
                  {
                    "name": "name",
                    "type": "string"
                  },
                  {
                    "name": "type",
                    "type": "string"
                  },
                  {
                    "name": "index",
                    "type": "int"
                  }
                ]
              }
            }
          },
          {
            "name": "pkkey",
            "type": {
              "type": "array",
              "items": {
                "name": "PkKey",
                "type": "record",
                "fields": [
                  {
                    "name": "columnname",
                    "type": "string"
                  },
                  {
                    "name": "value",
                    "type": "string"
                  }
                ]
              }
            }
          },
          {
            "name": "transaction",
            "type": {
              "name": "Transaction",
              "type": "record",
              "fields": [
                {
                  "name": "commitlsn",
                  "type": "string"
                },
                {
                  "name": "beginlsn",
                  "type": "string"
                },
                {
                  "name": "sequencenumber",
                  "type": "int"
                },
                {
                  "name": "finalevent",
                  "type": "boolean"
                },
                {
                  "name": "committime",
                  "type": "string"
                }
              ]
            }
          }
        ]
      }
    },
    {
      "name": "eventrow",
      "type": {
        "name": "EventRow",
        "type": "record",
        "fields": [
          {
            "name": "old",
            "type": "string"
          },
          {
            "name": "current",
            "type": "string"
          }
        ]
      }
    }
  ]
}

Contenuti correlati

• Che cos'è lo streaming di eventi di modifica?
• Configurare lo streaming di eventi di modifica
• Domande frequenti su CES