FractalDataWorks.Etl.Abstractions 0.4.0-preview.6

FractalDataWorks.Etl.Abstractions

ETL (Extract, Transform, Load) abstractions for scheduling and monitoring.

Overview

This package provides TypeCollections and interfaces for ETL scheduling and monitoring within the FractalDataWorks orchestration framework.

Key Types

Scheduling

• IEtlSchedulingService - Service for scheduling ETL pipeline executions
• IJobExecutor - Interface for job execution
• IScheduleInfo / IScheduleRequest - Schedule definitions

TypeCollections:

• ScheduleTypes - One-time vs Recurring schedules
• RecurringScheduleTypes - Daily, Weekly, Monthly, Cron, Interval
• ScheduleStatuses - Active, Paused, Completed, Error, Expired, Disabled
• SchedulePriorities - Critical, High, Normal, Low

Monitoring

• IEtlMetricsCollector - Collect ETL metrics
• IEtlTelemetryService - Telemetry and tracing
• IHealthCheckService - Health monitoring
• IAlertingService - Alert notifications

TypeCollections:

• HealthStates - Healthy, Degraded, Unhealthy
• AlertSeverities - Info, Warning, Error, Critical
• SeverityLevels - Verbose, Information, Warning, Error, Critical
• ComparisonOperators - Equal, NotEqual, GreaterThan, LessThan, etc.

Usage

Scheduling a Pipeline

From IEtlSchedulingService.cs:11-31:

public interface IEtlSchedulingService
{
    Task<IGenericResult<string>> ScheduleOnce(
        IScheduleRequest request,
        CancellationToken cancellationToken = default);

    Task<IGenericResult<string>> ScheduleRecurring(
        IRecurringScheduleRequest request,
        CancellationToken cancellationToken = default);

    Task<IGenericResult> PauseSchedule(
        string scheduleId,
        CancellationToken cancellationToken = default);

    Task<IGenericResult> ResumeSchedule(
        string scheduleId,
        CancellationToken cancellationToken = default);
}

From IScheduleRequest.cs:11-57:

public interface IScheduleRequest
{
    string PipelineId { get; }
    string ScheduleName { get; }
    string? Description { get; }
    DateTimeOffset? ExecuteAt { get; }
    IReadOnlyDictionary<string, object?> Parameters { get; }
    ISchedulePriority Priority { get; }
    string TimeZone { get; }
    bool IsEnabled { get; }
    IReadOnlyList<string> Tags { get; }
}

The concrete ScheduleRequest class is available in FractalDataWorks.Etl.Scheduling and uses a builder pattern. See that package for implementation details.

Checking Health

From IHealthCheckService.cs:13-28:

public interface IHealthCheckService
{
    Task<IGenericResult<IHealthStatus>> CheckHealth(CancellationToken cancellationToken = default);

    void RegisterHealthCheck(string name, Func<CancellationToken, Task<IHealthCheckResult>> check);
}

From IHealthCheckService.cs:33-49:

public interface IHealthStatus
{
    IHealthState Status { get; }
    IReadOnlyDictionary<string, IHealthCheckResult> Checks { get; }
    TimeSpan TotalDuration { get; }
}

From IHealthState.cs:9-20:

public interface IHealthState : ITypeOption<int, HealthStateBase>
{
    bool IsHealthy { get; }
    bool RequiresAttention { get; }
}

Metrics Collection

From IEtlMetricsCollector.cs:9-129:

public interface IEtlMetricsCollector
{
    void RecordPipelineStart(string executionId, string pipelineId, IDictionary<string, string>? properties = null);

    void RecordPipelineComplete(
        string executionId,
        string pipelineId,
        bool success,
        TimeSpan duration,
        IDictionary<string, string>? properties = null);

    void RecordRecordsProcessed(string executionId, string stageId, long count);

    void RecordThroughput(string executionId, string stageId, double recordsPerSecond);

    void RecordError(
        string executionId,
        string stageId,
        string errorType,
        string errorMessage,
        IDictionary<string, string>? properties = null);
}

TypeCollections

From ScheduleTypes.cs:14-17:

[TypeCollection(typeof(ScheduleTypeBase), typeof(IScheduleType), typeof(ScheduleTypes))]
public sealed partial class ScheduleTypes : TypeCollectionBase<ScheduleTypeBase, IScheduleType>
{
}

From HealthStates.cs:14-17:

[TypeCollection(typeof(HealthStateBase), typeof(IHealthState), typeof(HealthStates))]
public sealed partial class HealthStates : TypeCollectionBase<HealthStateBase, IHealthState>
{
}

TypeOption example from UnhealthyState.cs:11-18:

[TypeOption(typeof(HealthStates), "Unhealthy")]
public sealed class UnhealthyState : HealthStateBase
{
    public UnhealthyState() : base(2, "Unhealthy", isHealthy: false, requiresAttention: true) { }
}

Dependencies

• FractalDataWorks.Collections - TypeCollection infrastructure
• FractalDataWorks.Results - Railway-oriented result types

Related Packages

• FractalDataWorks.Etl.Scheduling - Concrete scheduling implementations