When data arrives late: monitoring freshness and recovering pipelines

Data Engineering / DataOps / ETL Monitoring / Data Reliability

Why a Running ETL Doesn’t Always Mean Up-to-Date Data is Available

Corporate information systems are increasingly dependent on regular data transmission and processing.

For example, ERP records business transactions, CRM maintains customer directories, and the electronic document management system stores relevant documents.

Based on these sources, the Data Warehouse generates management reports.

ETL, ELT, Change Data Capture, APIs, and streaming integrations are used for this purpose.

In large enterprises, hundreds of independent information processes can run daily.

Some load financial documents. Others update MDM. Still others create analytical data marts.

However, an increase in the number of integration procedures creates a new problem — ensuring the reliability of the entire information chain. For example, the nightly ETL completes without technical errors, but a part of the ERP did not transmit information.

Alternatively, the process successfully updated the DWH table but used data that was two hours old instead of current data. Another scenario is that a Spark Job completed its calculations, but the results were not fully published to the target data warehouse.

For the user, all these situations have a common consequence: the corporate report does not reflect the expected business state. At the same time, technical monitoring might indicate that servers are running, the DBMS is available, and most jobs have a ‘Success’ status.

Therefore, the reliability of a Data Pipeline cannot be determined solely by the fact of successful code execution. It is also necessary to monitor completeness, timeliness, compliance with business rules, and the ability to recover results after an error.

This is precisely why Data Freshness, Service Level Agreements, systematic monitoring, and managed Recovery procedures are becoming increasingly important for corporate Data Engineering.

What is a Data Pipeline

A Data Pipeline is a sequence or graph of operations that ensure the acquisition, transformation, and transfer of information. For example, a corporate sales process might consist of several stages.

• ERP / CRM / ECM
• Initial Information Sources
• Extract / CDC / API
• Change Acquisition
• Staging / Raw
• Intermediate Storage
• Data Quality / MDM
• Verification and Reconciliation
• ETL / ELT / Spark
• Transformations
• Data Warehouse
• Prepared Set
• BI / Data Mart
• Available Business Outcome

Simplified Data Pipeline. Reliability must be checked at each stage, not just after the completion of the last task. Each component can have its own technical status.

For example, the API returned a successful response. ETL completed without error.

Spark executed SQL transformation. However, the overall correctness of the data depends on the coordinated work of all stages.

Therefore, it is advisable to introduce end-to-end reliability metrics.

Five characteristics of a reliable information pipeline

For a corporate Data Pipeline, five main characteristics can be defined.

These characteristics must be evaluated independently. For example, the DWH may be available but contain outdated records.

Or information may arrive quickly, but with a large number of errors. Another case is when the results are correct, but the process takes so long that the analytics become available only after the relevant business decision has been made.

Therefore, for each critical dataset, requirements for all key characteristics must be defined.

Data Freshness: How Current is Corporate Data

Data Freshness describes the timeliness of information. For example, a company uses a real-time order dashboard.

The business expects new orders to appear on it within minutes. If the actual delay is two hours, the system does not meet expectations, even if all records are eventually loaded correctly.

At the same time, such a delay might not matter for financial reporting for the previous month. Therefore, Freshness must be defined according to the specific business task.

Event Time

Actual time of the business event.

Source Commit Time

Moment of operation confirmation in the original system, if such a moment is available.

Ingestion Time

Time of information reception by the integration platform.

Processing Time

Time of execution of the corresponding transformation.

Availability Time

The moment when the result becomes available to the defined consumer. These values can differ significantly.

For example, the order was created at 14:00, sent to Kafka at 14:01, processed by Spark at 14:03, and published to DWH at 14:05. In this case, the end-to-end latency from creation to availability is five minutes.

However, for another process, it is more appropriate to measure latency from the confirmation of the ERP transaction, rather than from the initial creation of the order draft. Therefore, it is necessary to clearly define the start and end points of measurement.

Why MAX(updated_at) doesn’t always show true freshness

One common way to assess recency is to check the maximum timestamp.

For example:

SELECT
    MAX(updated_at) AS latest_update
FROM dwh.fact_sales;

This can be a useful control indicator. However, it has limitations.

Suppose a company received only one new record from the ERP. Its update time is 07:55.

A fresh timestamp will appear in the DWH table. However, the rest of the documents from the last hour might not have made it to the data warehouse.

Therefore, MAX(updated_at) will show the current time, even though the dataset is incomplete. Another problem arises if the field is updated during ETL, rather than reflecting the time of the initial business event.

In this case, the check might show the time of the technical load. Therefore, for correct Freshness Monitoring, it is advisable to combine several characteristics.

For example, the time of the last successful load, the source control limit, the number of records received, and the results of relevant reconciliations.

Audit Table

For a corporate DWH, a separate registry can be created:

CREATE TABLE etl_batch_audit (
    batch_id          VARCHAR(100),
    source_system     VARCHAR(100),
    started_at        TIMESTAMP,
    completed_at      TIMESTAMP,
    source_watermark  TIMESTAMP,
    source_count      BIGINT,
    loaded_count      BIGINT,
    status            VARCHAR(30)
);

This is a simplified SQL example, not a universally compatible DDL for all DBMS. For example, SQL Server uses its own date and time types, including datetime2.

For a productive implementation, it is also necessary to define unique keys, transactional rules, and status update procedures. The main thing is to maintain a controlled state for each load.

SLA, SLO, and SLI for Enterprise Data

In traditional IT infrastructure, a Service Level Agreement often defines the acceptable service availability level. It is advisable to use a similar approach for information pipelines.

However, it is necessary to distinguish between SLA, SLO, and SLI.

Service Level Indicator

SLI is a specific measurable indicator.

For example, the delay in DWH updates or the proportion of successfully completed scheduled loads.

Service Level Objective

SLO is the target value of a metric.

For example, for a defined set of enterprises, the target is set to update it by 08:00 every business day.

Service Level Agreement

SLA is a formalized agreement regarding the corresponding service level and the obligations of the parties. An internal technical target is not necessarily a contractual SLA.

This is an important distinction.

Example Requirements

These are conditional goals, not universal standards. Actual requirements should be determined by the business.

For example, for financial reporting, correctness may have a higher priority than minimal delay. Conversely, for operational monitoring, timeliness is important, and certain indicators may be preliminary before the full reconciliation is completed.

Therefore, it is also advisable to indicate the status of the information set.

Importance of Control Windows and Business Calendar

A regular technical schedule is not sufficient for corporate information processes. For example, financial ETL must take into account month-end closing.

For some companies, the relevant process can only be executed after specific accounting operations are completed.

Therefore, the Data Pipeline must consider the business calendar.

Working and Non-Working Days

Not all sources are updated daily.

Period Closing

Financial reporting may have a special schedule.

Dependencies

Fact updates may depend on MDM completion.

Publication

The analytical data mart should become available after the necessary Quality Gates.

Escalation

If the process is not completed by the set time, responsible employees must be informed.

Thus, the calendar model must be part of the information contract.

Apache Airflow 1.10.2: Orchestration and Execution Control

Apache Airflow can be used to manage enterprise Data Pipelines. In January 2019, the project became a Top-Level Project of the Apache Software Foundation.

Also in January, Airflow 1.10.2 was released, which is already available at the date of this article.

The platform allows defining DAGs, dependencies, retries, and corresponding deadline control mechanisms.

SLA for Tasks

In Airflow, SLAs can be defined at the level of the corresponding task.

For example:

from datetime import timedeltaload_task = BashOperator(    task_id='load_finance_mart',    bash_command='/opt/etl/load_finance.sh',    sla=timedelta(hours=2),    dag=dag)

This is a simplified example of Airflow 1.10 syntax. It demonstrates the ability to define an appropriate time criterion.

However, it is important to correctly understand the semantics of Airflow SLA. It is related to the logical run period and the DAG schedule, and does not necessarily mean measuring two hours from the actual task start time.

Additionally, an SLA Miss does not guarantee automatic termination of execution. It is a mechanism for monitoring deadlines and notifying about their violation.

For a full business SLA, it is necessary to separately define when the information set became available to the end consumer.

Dependencies

Airflow allows not running the following tasks until the necessary prerequisites are met.

Retries

For temporary technical failures, retries can be applied.

Monitoring

Through the web interface, statuses and logs can be analyzed.

Limitations

Airflow does not automatically determine the correctness of financial indicators.

Therefore, specialized Data Quality procedures are needed.

Apache NiFi 1.9.0: New Features for Data Flow Control

Apache NiFi 1.9.0 was released on February 19, 2019. This is an important event for enterprises using NiFi to ingest, route, and process enterprise information.

Among the innovations are the ability to load new processors and controller services without stopping nodes, development of work with Kerberos-secured databases, improvements to Record Readers and Writers, and additional security headers for the web interface.

For enterprise Data Pipelines, these changes are of practical importance.

Dynamic Component Loading

New processors can be made available without the need for a full node stop for relevant scenarios. However, this does not mean that any NiFi update or any flow change can be performed without risk and downtime.

Compatibility and dependencies of specific components must be considered.

Kerberos-secured DBMS

The development of support for relevant connections is important for enterprises with centralized authentication mechanisms.

Record Readers and Writers

Improving automatic schema detection can simplify working with structured messages. However, automatic type detection should not replace an approved information contract.

For example, the system might incorrectly interpret a numeric identifier as a regular number, even though leading zeros are important for the business.

Therefore, for critical sources, it is better to use defined schemas and checks.

Information Security

NiFi 1.9 adds support for a number of standard HTTP security headers. However, they are only part of the overall security model.

The enterprise also needs to configure authentication, authorization, TLS, and access control to the relevant resources.

Data Provenance in NiFi

One of NiFi’s important features is Data Provenance. It allows logging the lifecycle events of a FlowFile.

For example, one can establish when a file was received, which processors it passed through, and what corresponding actions were performed.

This is useful for investigating technical issues.

Example

An enterprise receives a CSV file from a partner system. NiFi performs format validation and transfers the records to PostgreSQL.

If part of the information did not load, Data Provenance can help trace the relevant operations. However, limitations should be considered.

Data Provenance is not a universal mechanism for complete enterprise Data Lineage at the business KPI level. It does not necessarily explain the formula for calculating financial revenue in the DWH.

This requires additional metadata and business descriptions.

Retention

The provenance history also has a defined retention period.

Therefore, it should be taken into account that old events may become unavailable after the corresponding period expires.

Protection

Metadata and logs may contain confidential information.

Therefore, access to them must be controlled.

Backpressure: protecting the information pipeline from overload

One practical problem is a situation where the source generates information faster than the consumer can process it. For example, an ERP exports a significant volume of documents.

The target DBMS is temporarily slow. As a result, messages accumulate between components.

If the system does not control the queues, memory and disk usage increases.

Apache NiFi supports Back Pressure mechanisms on the corresponding connections between processors.

By number of FlowFiles

A threshold for the number of objects in the queue can be defined.

By data volume

It is possible to set limits on the total size.

Practical result

If the next processor cannot keep up with processing information, the upstream component can temporarily stop further transmission through a filled connection. However, backpressure does not eliminate the cause of slow processing.

If the target DBMS systematically fails to cope with the load, it is necessary to review the architecture, performance, or regulations. It is also important to monitor whether queues are accumulating over a long period of time.

Kafka Consumer Lag: streaming consumer delay

For event-driven integration, one of the important metrics is Consumer Lag. For example, ERP publishes orders to Kafka.

A separate consumer transmits them to the analytical system. If messages arrive faster than they are processed, the lag increases.

This can mean that the Data Warehouse gradually loses relevance.

What to Monitor

The number of unprocessed messages relative to the corresponding log position.

Why Lag Alone is Insufficient

A lag of a thousand messages can mean seconds or hours depending on the stream intensity and processing performance.

Therefore, it is also necessary to assess the actual time delay.

Retention

If the lag increases beyond the established message retention period, part of the history may be lost for a specific consumer.

Recovery

After recovery, the consumer can re-receive relevant messages depending on the committed progress.

Therefore, idempotency is required.

Business Readiness

Even zero Consumer Lag does not guarantee that the corresponding DWH has already successfully published verified analytical results.

Therefore, it is necessary to control the entire information chain.

Error Handling: Technical and Business Exceptions

For corporate Data Pipelines, it is important to distinguish between error types. For example, SQL Server is temporarily unavailable.

This is a technical problem that can be resolved by retrying. Another case is when the corporate vendor code is missing in the document.

A re-run will not create the correct code automatically. This type of violation requires a proper Data Quality procedure.

It is important not to use infinite retries for all error types. For example, if the API structure changed in an incompatible way, ten thousand retries will not solve the problem.

They will only create additional load. Therefore, exception classification is needed.

Dead Letter Queue and Quarantine Zone

Separate controlled queues or storage can be used for incorrect messages. For example, in Kafka, a special topic for problematic events can be organized at the application architecture level.

Important: a standard Kafka broker does not automatically create a universal Dead Letter Topic for any consumer. This requires appropriate implementation or a supported connector.

Purpose

Do not lose information about a message that cannot be processed correctly.

Metadata

It is advisable to store the event identifier, source, timestamp, and violation code.

Data Protection

The full content of personal messages should not be unnecessarily duplicated.

Reprocessing

After the cause has been eliminated, the message can be reprocessed.

Lifecycle Management

The Quarantine Zone should not become a permanent storage for uncorrected records.

Each category requires responsible parties and defined lifecycle rules.

Idempotency and Failure Recovery

One of the most important characteristics is correct behavior after repeated execution. For example, ETL received a thousand documents and loaded them into DWH.

After that, the process crashed before confirming success. During the next run, it may re-receive the same documents.

If a standard uncontrolled INSERT is used, duplication is possible. Therefore, an idempotent mechanism must be created.

Business Keys

For business documents, the correct combination of unique attributes is defined.

For example, source, legal entity, document type, and number, where such a combination corresponds to the actual rules.

Staging

Initial records are loaded into a controlled staging area.

Merge / Upsert

Depending on the DBMS capabilities, insert or update mechanisms based on a defined key are used.

Transactions

It is necessary to correctly coordinate data recording and confirmation of the processed batch.

Batch Audit

The status of the corresponding load is recorded.

Recovery

After recovery, the process determines which operations need to be repeated. It is important not to assume that the mere presence of a Batch ID guarantees exactly-once.

This requires a proper transactional implementation. Operations spanning multiple independent systems can be particularly complex.

Data Quality Gates: The Final Check Before Publication

For a reliable enterprise DWH, it’s important to separate the technical completion of loading from the business readiness of the dataset. For example, the ETL received data from three ERPs.

SQL transformations are complete. However, one ERP transmitted only a portion of the documents.

If the financial data mart is published without verification, management will receive an incomplete result. Therefore, Quality Gates must be used.

Source Completeness

All mandatory sources must provide the required datasets.

Record Counts

The record count must comply with established control rules.

MDM Mapping

Critical reference matches must be confirmed.

Reconciliation

Financial totals must align with the approved methodology.

Historical Consistency

For SCD, the correctness of historical intervals must be verified.

Data Freshness

The dataset must meet the established time criterion. Only after passing the necessary checks can the information receive the appropriate readiness status.

For research datasets, the criteria may be softer. For official financial reporting, they are stricter.

Therefore, the rules should be established by business owners.

Disaster Recovery for Information Pipelines

Traditional Disaster Recovery often focuses on restoring servers and DBMS.

However, for modern analytics, it is also necessary to restore information processes. For example, after a failure, the company restored the SQL Server DWH.

However, the Kafka consumer began to retransmit some of the previous messages. Or Airflow recovered with old metadata state.

As a result, analytical datasets may become inconsistent. Therefore, several components need to be considered.

Importantly, successful recovery of all servers does not necessarily guarantee data correctness.

Control reconciliations are needed after Recovery.

RPO and RTO

For each critical process, target indicators for acceptable information loss and recovery time are defined.

Replay

Under appropriate conditions, stored messages can be reprocessed.

Backfill

For batch processes, reprocessing of defined historical periods can be performed.

Reconciliation

After recovery, the quantity and correctness of records must be verified.

Thus, Disaster Recovery becomes part of the overall information architecture.

Practical Scenario: Nightly ETL Recovery for a Trading Holding

Let’s consider a hypothetical enterprise that uses three ERPs, a centralized MDM, and a corporate DWH.

The daily load must complete by 08:00.

One morning, the team discovers that the data for one subsidiary has not been updated.

Stage 1. Automatic Detection

Monitoring identifies that the control time has passed and the relevant dataset has not reached a ready status.

Stage 2. Incident Registration

A ticket is created with the appropriate priority.

Stage 3. Source Verification

It is determined whether the ERP is available and if the necessary information has been generated.

Stage 4. ETL Analysis

Airflow logs or logs from another platform are checked.

Stage 5. Error Classification

It is determined whether the issue is technical or data-related.

Stage 6. Recovery

Under appropriate conditions, the necessary stage is re-executed.

Stage 7. Idempotency

It is checked that previously loaded records are not duplicated.

Stage 8. MDM

Compliance with corporate directories is monitored.

Stage 9. Reconciliation

Financial summaries are reconciled with authoritative sources.

Stage 10. Publication

After successful checks, the dataset receives the appropriate status.

Stage 11. Communication

Users receive information about the actual availability of the report.

Stage 12. Post-Incident Review

The team establishes the cause and determines preventive measures. For example, if the issue arose due to temporary unavailability of the ERP, a review of the retry policy may be appropriate.

If the cause is related to a new document status, a change to the relevant information contract is needed.

Thus, Recovery must include not only a restart but also a result verification.

DataOps and Continuous Reliability Improvement

• The reliability of information flows cannot be ensured by a one-time setup.
• Corporate sources are changing.
• Data volumes are growing.
• New information contracts are appearing.
• Therefore, it is necessary to constantly analyze the actual performance of pipelines.
• DataOps practices are useful for this.

Version Control

Changes to SQL, DAGs, and integration configurations are stored in Git.

Automated Testing

For critical transformations, corresponding tests are performed.

Monitoring

Freshness, Completeness, Data Quality, and technical specifications are monitored.

Incident Management

Errors are assigned to designated responsible parties.

Root Cause Analysis

After a significant incident, the root cause is established.

Continuous Improvement

Investigation results are used to improve rules and tests. For example, if a new ERP status caused an incorrect financial report, it’s necessary not only to fix the SQL but also to add a corresponding Regression Test.

This reduces the risk of a similar problem recurring.

Key Performance Indicators (KPIs) for Data Pipeline Reliability

It is advisable for an enterprise to establish a system of metrics.

It is important to avoid formal evaluation. For example, Pipeline Success Rate can be 99%, but a single failed processing might concern the most critical financial data mart.

Therefore, the business criticality of processes must be taken into account. Similarly, a high DWH availability level does not guarantee data relevance.

Thus, metrics need to be evaluated comprehensively.

Information Security and GDPR

For enterprises subject to GDPR, a Data Pipeline can be part of personal data processing. For example, CRM transfers relevant records to DWH.

During technical errors, such data may end up in the Quarantine Zone or logs.

Therefore, it is necessary to establish appropriate rules.

Minimization

Full personal records should not be unnecessarily duplicated in diagnostic messages.

Access Control

Technical accounts must have the minimum necessary privileges.

Retention

Defined retention periods are required for intermediate and problem sets.

Data Lineage

Relevant data flows need to be documented.

Incident Management

In case of a confirmed personal data security breach, appropriate assessment and notification procedures must be applied. However, a normal technical ETL error is not necessarily a personal data breach.

The nature of the event and its consequences must be assessed separately.

Therefore, technical monitoring and legal obligations should not be equated.

Intecracy Group Competencies in Ensuring Data Pipeline Reliability

Building reliable data pipelines requires a combination of several technological areas.

At the same time, the criteria for relevance, completeness, and business correctness must be defined by the respective owners of the enterprise’s information domains.

Practical Guidelines for Implementation

The reliability of Data Pipelines is becoming an important component of the corporate analytical architecture. For enterprises, it is not enough to control only the server status or the fact of successful ETL completion.

It is necessary to measure Data Freshness, information completeness, correctness of business indicators, and the ability to recover processes after failures.

Apache Airflow 1.10 allows organizing dependent tasks and corresponding control mechanisms. Apache NiFi 1.9.0, released in February 2019, enhances the capabilities of corporate stream integration.

Apache Kafka and Spark Structured Streaming offer tools for event processing with corresponding mechanisms for progress saving and recovery. However, no platform automatically guarantees exactly-once for all external operations or complete correctness of corporate data.

This requires idempotency, transactional logic, Data Quality, MDM, and control reconciliations.

Properly defined SLAs and SLOs, reflecting actual business needs, are of particular importance.

The main principle of a reliable corporate Data Pipeline is that the information process should not just complete without a technical error, but should timely provide complete, correct, and verified data – with the ability to explain every significant deviation and restore the result after a failure.