Enriching event streams with entity data

Patent metadata

Field	Value
Publication number	US-12547639-B2
Application number	US-202418808541-A
Country	US
Kind code	B2
Filing date	Aug 19, 2024
Priority date	Aug 22, 2023
Publication date	Feb 10, 2026
Grant date	Feb 10, 2026

Abstract

System and method for enriching a data stream with enrichment data. The system loads data from one or more customer data warehouses into a storage component using an ingest pipeline; receives, at an enrichment pipeline, an incoming data stream; determines, using the enrichment pipeline, an insertion point within the incoming data stream, the insertion point corresponding to a data object mention; determines, using the enrichment pipeline, enrichment data matching the data object mention, the enrichment data being retrieved from the storage component; augments, via the enrichment pipeline, the incoming data stream with the enrichment data at the determined insertion point to generate an enriched data stream, and transmits the enriched data stream to one or more destinations. The data stream can be an event stream. The enrichment data can be entity data. The system can use a reverse extract/transform/load (ETL) model to enable data ingesting and/or data stream enrichment.

First claim

What is claimed is:

1 . A system comprising:
one or more computer processors; one or more computer memories; and a set of instructions stored in the one or more computer memories, the set of instructions configuring the one or more computer processors to perform operations, the operations comprising: loading data from one or more customer data warehouses into a storage component using an ingest pipeline, the loading of the data comprising:
creating a source corresponding to a table in a customer data warehouse of the one or more customer data warehouses; and
creating a reverse extract-transform-load (ETL) model associated with the source, the reverse ETL model comprising a relational query comprising table information for the table, the reverse ETL model remaining unchanged upon detecting an update to a schema of the table;
receiving, at an enrichment pipeline, a data stream; determining, by the enrichment pipeline, an insertion point within the data stream, the insertion point corresponding to a data object mention; determining, by the enrichment pipeline, enrichment data matching the data object mention, the enrichment data being retrieved from the storage component; augmenting, using the enrichment pipeline, the data stream with the enrichment data at the determined insertion point to generate an enriched data stream; and transmitting the enriched data stream to one or more destinations.

2 . The system of claim 1 , wherein the ingest pipeline comprises a scheduler component that determines at least one of a timing or a frequency of data synchronization operations between the one or more customer data warehouses and the storage component.

3 . The system of claim 2 , wherein the ingest pipeline comprises a loader component, the loader component enabled to:
receive, from the scheduler component, synchronization information corresponding to a first job to be executed as part of an data ingest task; upon receiving the synchronization information associated with the first job, create a second job for a data processing engine based on the first job, the second job being associated with an application programming interface (API) to the storage component; and execute the second job, the executing of the second job comprising one of at least a data write operation, data retrieval or data deletion operation associated with the storage component.

4 . The system of claim 1 , wherein:
the enrichment pipeline uses a data processing engine associated with an execution plan; and upon receiving an incoming data stream and detecting that a downstream component is configured to receive an enriched data stream, adding an execution graph node to the execution plan for the data processing engine, the execution graph node associated with a call to an enrichment endpoint for an API to the storage component.

5 . The system of claim 4 , wherein:
the data object mention corresponds to an entity ID; and determining the insertion point within the incoming data stream comprises detecting the entity ID in the incoming data stream using a path rule and the enrichment endpoint for the API associated with the storage component.

6 . The system of claim 5 , wherein:
the enrichment data comprises one or more entity attributes, each entity attribute associated with at least one attribute value; the enrichment data matching the data object mention comprises an entity attribute of the one or more entity attributes matching the entity ID based on a matching criterion; and the enrichment data is retrieved from the storage component using the API.

7 . The system of claim 1 , wherein the data stream corresponds to an event stream and the storage component corresponds to a cache component.

8 . The system of claim 1 , wherein:
the source and the reverse ETL model are associated with an entity model; and the reverse ETL model further comprises synchronization schedule information associated with the table corresponding to the source.

9 . The system of claim 8 , the operations further comprising creating a mapping between the reverse ETL model and a destination of the one or more destinations.

10 . The system of claim 9 , the operations further comprising:
displaying one of at least the reverse ETL model, the entity model and the mapping between the reverse ETL model and the destination in a user interface (UI); upon receiving user input indicative of a synchronization failure associated with the source or of a revision to the synchronization schedule information, updating the reverse ETL model; and upon receiving user input indicative of a revision to the mapping between the reverse ETL model and the destination, updating the mapping.

11 . A method comprising:
loading data from one or more customer data warehouses into a storage component using an ingest pipeline, the loading of the data comprising:
creating a source corresponding to a table in a customer data warehouse of the one or more customer data warehouses; and
creating a reverse extract-transform-load (ETL) model associated with the source, the reverse ETL model comprising a relational query comprising table information for the table, the reverse ETL model remaining unchanged upon detecting an update to a schema of the table;
receiving, at an enrichment pipeline, a data stream; determining, by the enrichment pipeline, an insertion point within the data stream, the insertion point corresponding to a data object mention; determining, by the enrichment pipeline, enrichment data matching the data object mention, the enrichment data being retrieved from the storage component; augmenting, using the enrichment pipeline, the data stream with the enrichment data at the determined insertion point to generate an enriched data stream; and transmitting the enriched data stream to one or more destinations.

12 . The method of claim 11 , wherein the ingest pipeline comprises a scheduler component that determines at least one of a timing or a frequency of data synchronization operations between the one or more customer data warehouses and the storage component.

13 . The method of claim 12 , wherein the ingest pipeline comprises a loader component, the loader component enabled to:
receive, from the scheduler component, synchronization information corresponding to a first job to be executed as part of an data ingest task; upon receiving the synchronization information associated with the first job, creating a second job for a data processing engine based on the first job, the second job being associated with an application programming interface (API) to the storage component; and execute the second job, the executing of the second job comprising one of at least a data write operation, data retrieval or data deletion operation associated with the storage component.

14 . The method of claim 11 , wherein:
the enrichment pipeline uses a data processing engine associated with an execution plan; and upon receiving an incoming data stream and detecting that a downstream component is configured to receive an enriched data stream, adding an execution graph node to the execution plan for the data processing engine, the execution graph node associated with a call to an enrichment endpoint for an API to the storage component.

15 . The method of claim 14 , wherein the data object mention corresponds to an entity ID, and determining the insertion point within the incoming data stream comprises detecting the entity ID in the incoming data stream using a path rule and the enrichment endpoint for the API associated with the storage component.

Assignees

• Twilio Inc

Inventors

• Lasky Daniel Joseph
• Ishikawa Akihiro
• Thompson Drew
• Anderson Jon
• Mehta Udit

Classifications

• G06F16/27

  Replication, distribution or synchronisation of data between databases or within a distributed database system; Distributed database system architectures therefor · CPC title

• G06F16/283

  Multi-dimensional databases or data warehouses, e.g. MOLAP or ROLAP · CPC title

• G06F16/254 Primary

  Extract, transform and load [ETL] procedures, e.g. ETL data flows in data warehouses · CPC title

• G06F16/24568 Primary

  Data stream processing; Continuous queries · CPC title