Legacy Pipeline Atlas worker-1snapshot 2026-07-15

MDClarity · task primitive

ProcessDataFile

Overview

ProcessDataFileTask scans a delimited data file to infer its structure, imports the file contents into an import table, and then merges the imported data into a source table. This task combines file scanning, loading, and merge operations into a single workflow, making it ideal for files with unknown or dynamic structures.

Use this task when: You need to import files without pre-configured file specifications, want automatic structure detection, or need intelligent merge logic to synchronize external data into source tables.

Don't use this task when: You already have a known file structure and can use LoadDataFileTask for better performance, need to load to non-source tables, or require complex transformations beyond simple merge operations.

Parameters

Parameter Type Required Default Description
FilePath string Yes None Path to the delimited data file to import. Can be a local filesystem path or S3 URI.
BaseTableName string Yes None Base name for generated tables. The task creates Import{BaseTableName} and Source{BaseTableName} tables automatically.
RowTerminator string No \r\n Character(s) that mark the end of each row. Common values: \r\n (Windows), \n (Unix), or custom terminators.
ColumnSeparator string No , Character(s) that separate columns. Common values: , (comma), `
TextQualified bool No true Whether column values are wrapped in quotes. When true, the parser handles quoted values that may contain the column separator.
HasHeader bool No true Whether the first row contains column names. When true, column names are read from the file; when false, columns are named Column1, Column2, etc.
MaxInvalidRows int No 0 Maximum number of invalid rows to tolerate during import before failing. 0 means fail on first invalid row.
BatchSize int No 5000 Number of rows to load per batch during bulk copy operations. Higher values = faster import but more memory usage.
ArchiveDirectory string No None Optional directory to archive the source file after successful import. File is moved with a timestamp appended to the filename. If empty/null, file is not archived.
Columns List<ColumnSpec> No Auto-scanned Column specifications for the file. Usually auto-populated by the CreateDataFileSpec action, but can be manually specified.

Execution Flow

flowchart TD
    A[Start ProcessDataFile] --> B[Create DataFileSpec from parameters]
    B --> C[Create DataStreamMergerFactory]
    C --> D[Create DataStreamMerger with spec and file path]
    D --> E[Call ImportToSource method]
    E --> F[Drop and recreate Import table]
    F --> G[Load file data into Import table]
    G --> H{Valid rows loaded?}
    H -->|No - exceeds MaxInvalidRows| Z[Throw Exception]
    H -->|Yes| I[Create clustered index on Import table keys]
    I --> J{Duplicate keys found?}
    J -->|Yes| Z
    J -->|No| K{ArchiveDirectory specified?}
    K -->|Yes| L[Move file to archive with timestamp]
    K -->|No| M[Skip archiving]
    L --> N[Create Source table if not exists]
    M --> N
    N --> O[Merge Import table into Source table]
    O --> P{Key match in Source?}
    P -->|Yes| Q[UPDATE matching rows]
    P -->|No| R[INSERT new rows]
    Q --> S[End]
    R --> S

    style F fill:#ffe1e1
    style G fill:#e1f5ff
    style I fill:#fff4e1
    style O fill:#e1f5ff
    style Q fill:#fff4e1
    style R fill:#fff4e1

Processing Algorithm

The ProcessDataFile task performs an import-to-source merge workflow that intelligently synchronizes file data with database tables:

Merge Logic

The merge operation uses key columns (marked with IsKey = true in ColumnSpec) to determine how to handle each row:

  1. Identify Keys: All columns marked as IsKey form a composite key for matching
  2. Match Rows: For each row in the Import table, check if the key exists in the Source table
  3. UPDATE if exists: When keys match, update all non-key columns in the Source table with values from Import table
  4. INSERT if new: When keys don't match, insert the entire row as a new record in the Source table

This ensures the Source table reflects the current state of imported data without duplicates.

Example Execution

Given a file with columns: ID (key), Name, Status

Initial Import:

File: ID|Name|Status
      1|Alice|Active
      2|Bob|Active

Result: Source table has 2 rows (inserts)

Second Import (with changes):

File: ID|Name|Status
      1|Alice|Inactive  ← Updated status
      3|Charlie|Active  ← New record

Result: Source table has 3 rows:

Row Count: Initial = 2, After second import = 3 (1 updated, 1 unchanged, 1 inserted)

Usage Examples

Example 1: Basic File Import with Auto-Scan

// Import a pipe-delimited file with header, auto-detecting structure
var task = new ProcessDataFileTask
{
    FilePath = @"C:\Data\customers.txt",
    BaseTableName = "Customer",
    ColumnSeparator = "|",
    RowTerminator = "\r\n",
    HasHeader = true,
    TextQualified = true,
};

// This creates ImportCustomer and SourceCustomer tables automatically
// Columns are auto-scanned from the file

Example 2: File Import with Archive

// Import CSV file and archive it after processing
var task = new ProcessDataFileTask
{
    FilePath = @"C:\Incoming\daily_sales.csv",
    BaseTableName = "DailySales",
    ColumnSeparator = ",",
    HasHeader = true,
    ArchiveDirectory = @"C:\Archive\",
    MaxInvalidRows = 10, // Tolerate up to 10 bad rows
};

// After import, file is moved to C:\Archive\daily_sales_[id]_143022.csv

Example 3: Tab-Delimited File from S3

// Import tab-delimited file from S3 bucket
var task = new ProcessDataFileTask
{
    FilePath = "s3://mybucket/data/inventory.txt",
    BaseTableName = "Inventory",
    ColumnSeparator = "\t",
    RowTerminator = "\n",
    HasHeader = true,
    BatchSize = 10000, // Larger batches for big file
};

// Creates ImportInventory and SourceInventory tables
// Reads file directly from S3

Example 4: Pre-Configured Column Specs

// Import with manually specified column structure
var task = new ProcessDataFileTask
{
    FilePath = @"C:\Data\products.txt",
    BaseTableName = "Product",
    ColumnSeparator = "|",
    HasHeader = false, // No header row
    Columns = new List<ColumnSpec>
    {
        new ColumnSpec { Name = "ProductID", Type = ColumnType.Integer, IsKey = true },
        new ColumnSpec { Name = "ProductName", Type = ColumnType.String, Width = 200 },
        new ColumnSpec { Name = "Price", Type = ColumnType.Decimal },
    },
};

// Use this when you know the structure and want to skip auto-scan

Example 5: Unix Format File with No Text Qualifier

// Import Unix-formatted file without quoted text
var task = new ProcessDataFileTask
{
    FilePath = @"/data/logs/system.log",
    BaseTableName = "SystemLog",
    ColumnSeparator = "|",
    RowTerminator = "\n",
    TextQualified = false, // Values are not quoted
    HasHeader = true,
    MaxInvalidRows = 0, // Fail immediately on any invalid row
};

Related Tasks Comparison

Task Use Case Structure Detection Merge Logic Archive Support
ProcessDataFile Unknown structure, need auto-scan + merge Auto-scans file Yes (Import → Source) Yes (file archiving)
LoadDataFileTask Known structure, faster loading Uses pre-configured spec No (direct load) No
ScanDataFileTask Only scan file structure Auto-scans file No (scan only) No
ExecuteMergeSqlTask Custom SQL-based merge N/A (SQL query) Yes (Query → Table) No
ImportSqlTask Import from SQL Server/Snowflake N/A (SQL query) No (direct load) No

When to choose:

Performance Considerations

Database Impact

Optimization Tips

Scenario Recommendation Rationale
Large files (>1M rows) Increase BatchSize to 10,000-50,000 Reduces round trips, faster bulk copy
Wide rows (many columns) Decrease BatchSize to 1,000-2,000 Prevents memory pressure from large batches
Frequent imports Keep ArchiveDirectory on fast storage Archiving can bottleneck if slow disk/network
Slow merge performance Ensure key columns are indexed in Source table Merge uses keys for matching; indexes speed lookups
S3 files Run task in same AWS region as bucket Reduces network latency for file reads
High error tolerance Set MaxInvalidRows conservatively Too high = silent data quality issues

Concurrent Execution: Safe to run multiple ProcessDataFile tasks concurrently if they use different BaseTableName values. Concurrent imports to the same tables will cause deadlocks.

Error Handling

The ProcessDataFile task uses a fail-fast approach for data integrity issues:

// Key validation checks that cause immediate failure:
// 1. File cannot be read
// 2. Invalid rows exceed MaxInvalidRows threshold
// 3. Duplicate keys found in Import table
// 4. Merge operation fails (database error)

Important behaviors:

Error Messages:

Common Pitfalls

Issue Problem Solution
No key columns defined Without key columns, the merge operation cannot determine which rows to update vs insert Ensure at least one ColumnSpec has IsKey = true, or use the auto-scan feature which infers keys from file structure
Duplicate keys in source file If the file contains duplicate key values, the import will fail after loading to the Import table De-duplicate the source file before import, or modify key column selection to ensure uniqueness
File archived but merge failed The task archives files before merging, so a merge failure leaves the file archived but Source table incomplete Check archived file location and re-import manually, or disable ArchiveDirectory until merge is stable
Column spec mismatch Manually specified Columns don't match actual file structure, causing load failures Use the CreateDataFileSpec action to auto-scan first, or verify column specs match file exactly
MaxInvalidRows too high Setting this too high masks data quality issues, importing corrupt data Start with 0 (fail-fast), then increase cautiously if specific invalid rows are expected and acceptable
Import table exists before run The task expects to drop and recreate the Import table; manual table creation may cause conflicts Let the task manage the Import table lifecycle; don't create it manually
Source table schema drift If file structure changes (new columns), the Source table won't automatically adapt Drop the Source table to force recreation with new schema, or use ALTER TABLE manually
Large files with low BatchSize Default 5,000 batch size can make very large file imports slow Increase BatchSize to 25,000+ for files with millions of rows

Code References