Events and streams

Datastream organizes replicated data into a three-level hierarchy of streams, objects, and events. A stream connects a data source to a destination, while an object represents a specific source entity in that stream, such as a database table. An event captures a single data change from an object, such as a database insert.

Each event includes data and metadata that identify its source stream and object. You can use this metadata to track change origins, remove duplicate records in your destination, and maintain ordering across downstream consumers.

About events

Each event consists of three types of data:

• Event data: This represents the change to the data itself from the object originating from the stream source. Every event contains the entirety of the row that changed.
• Generic metadata: This metadata appears on every event generated by Datastream which is used for actions, such as removing duplicate data in the destination.
• Source-specific metadata: This metadata appears on every event generated by a specific stream source. This metadata varies by source.

Event data

Event data is the payload of every change from a given object originating from a stream source.

Events are in either the Avro or JSON format.

When working with the Avro format, for each column, the event contains the column index and value. Using the column index, the column name and unified type can be retrieved from the schema in the Avro header.

When working with the JSON format, for each column, the event contains the column name and value.

Event metadata can be used to collect information about the event's origin, as well as to remove duplicate data in the destination and order events by the downstream consumer.

The following tables list and describe the fields and data types for generic and source-specific event metadata.

Generic metadata

This metadata is consistent across streams of all types.

Source-specific metadata

This metadata is associated with CDC and backfill events from a source database. To view this metadata, select a source from the drop-down menu that follows.

Example of an event flow

This flow illustrates the events generated by three consecutive operations: INSERT, UPDATE, and DELETE, on a single row in a SAMPLE table for a source database.

INSERT (T0)

The message payload consists of the entirety of the new row.

{
  "stream_name": "projects/myProj/locations/myLoc/streams/Oracle-to-Source",
  "read_method": "oracle-cdc-logminer",
  "object": "SAMPLE.TBL",
  "uuid": "d7989206-380f-0e81-8056-240501101100",
  "read_timestamp": "2019-11-07T07:37:16.808Z",
  "source_timestamp": "2019-11-07T02:15:39",  
  "source_metadata": {
    "log_file": ""
    "scn": 15869116216871,
    "row_id": "AAAPwRAALAAMzMBABD",
    "is_deleted": false,
    "database": "DB1",
    "schema": "ROOT",
    "table": "SAMPLE"
    "change_type": "INSERT",
    "tx_id": 
    "rs_id": "0x0073c9.000a4e4c.01d0",
    "ssn": 67,
  },
  "payload": {
    "THIS_IS_MY_PK": "1231535353",
    "FIELD1": "foo",
    "FIELD2": "TLV",
  }
}

UPDATE (T1)

The message payload consists of the entirety of the new row. It doesn't include previous values.

{
  "stream_name": "projects/myProj/locations/myLoc/streams/Oracle-to-Source",
  "read_method": "oracle-cdc-logminer",
  "object": "SAMPLE.TBL",
  "uuid": "e6067366-1efc-0a10-a084-0d8701101101",
  "read_timestamp": "2019-11-07T07:37:18.808Z",
  "source_timestamp": "2019-11-07T02:17:39",  
  "source_metadata": {
    "log_file": 
    "scn": 15869150473224,
    "row_id": "AAAGYPAATAAPIC5AAB",
    "is_deleted": false,
    "database":
    "schema": "ROOT",
    "table": "SAMPLE"
    "change_type": "UPDATE",
    "tx_id":
    "rs_id": "0x006cf4.00056b26.0010",
    "ssn": 0,
  },
  "payload": {
    "THIS_IS_MY_PK": "1231535353",
    "FIELD1": null,
    "FIELD2": "TLV",
  }
}

DELETE (T2)

The message payload consists of the entirety of the new row.

{
  "stream_name": "projects/myProj/locations/myLoc/streams/Oracle-to-Source",
  "read_method": "oracle-cdc-logminer",
  "object": "SAMPLE.TBL",
  "uuid": "c504f4bc-0ffc-4a1a-84df-6aba382fa651",
  "read_timestamp": "2019-11-07T07:37:20.808Z",
  "source_timestamp": "2019-11-07T02:19:39",
  "source_metadata": {
    "log_file": 
    "scn": 158691504732555,
    "row_id": "AAAGYPAATAAPIC5AAC",
    "is_deleted": true,
    "database":
    "schema": "ROOT",
    "table": "SAMPLE"
    "change_type": "DELETE",
    "tx_id":
    "rs_id": "0x006cf4.00056b26.0011",
    "ssn": 0,
  },
  "payload": {
    "THIS_IS_MY_PK": "1231535353",
    "FIELD1": null,
    "FIELD2": "TLV",
  }
}

Ordering and consistency

This section covers how Datastream handles ordering and consistency.

Ordering

Datastream doesn't guarantee in-order event delivery, but each event contains the full row of data, the timestamp of when the data was written to the source, and ordering metadata.

Events that are part of an initial backfill have the read_method field set to backfill or ending with -backfill, -fulldump, or -incremental. Backfill events are generated out of order by design and can be consumed in any order.

Events that are part of ongoing change data capture (CDC) replication have the read_method field set to cdc or a source-specific CDC value, such as mysql-cdc-binlog, oracle-cdc-logminer, postgres-cdc-wal, sqlserver-cdc, or salesforce-cdc. How you order ongoing replication events across all sources depends on your destination:

Consistency

Datastream ensures that the data from the source database is delivered to the destination at least once. No event is missed, but there's a possibility of duplicate events in the stream. The window for duplicate events should be on the order of minutes, and the universally unique identifier (UUID) of the event in the event metadata can be used to detect duplicates.

However, don't rely solely on the UUID to deduplicate events downstream. If a pipeline interruption occurs, Datastream re-reads the source log from the last saved checkpoint to ensure delivery. Because Datastream dynamically generates UUIDs when it reads an event, this type of retry results in fresh UUIDs for the exact same source data. Additionally, during bulk operations in the source database, multiple distinct rows might share the same change sequence number (CSN).

To reliably deduplicate data, partition the events by the source table primary keys, and group them using the change sequence number (CSN) (the change_sequence_number field). This ensures that retries of the same event are correctly collapsed while preserving legitimately distinct row updates from bulk transactions.

When database log files contain uncommitted transactions, if any transactions are rolled back, then the database reflects this in the log files as "reverse" data manipulation language (DML) operations. For example, a rolled-back INSERT operation will have a corresponding DELETE operation. Datastream reads these operations from the log files.

About streams

Every stream has metadata that describes both the stream and the source from which it pulls data. This metadata includes information such as the stream name, the source and destination connection profiles.

To view the complete definition of the Stream object, see the API Reference documentation.

Stream state and status

A stream can be in one of the following states:

• Not started
• Starting
• Running
• Draining
• Paused
• Failed
• Failed permanently

You can use the logs to find additional status information, such as the tables backfilling or number of rows processed. You can also use the FetchStreamErrors API to retrieve errors.

Object metadata available using the discover API

The discover API returns objects which represent the structure of the objects defined in the data source or destination that's represented by the connection profile. Each object has metadata on the object itself, as well as for every field of data that it pulls. This metadata is available using the discover API.

What's next

• To learn more about streams, see Stream lifecycle.
• To learn how to create a stream, see Create a stream.