Bulk Output API - TypeScript

A TypeScript-based bulk data output API system with support for various data sources and processing workflows.

Developer setup

• Use pnpm workspaces. See the pnpm migration guide for install and daily commands.

⚠️ CRITICAL: Client Config Folder Naming Requirements

🚨 IMPORTANT: Deployment and runtime operations WILL FAIL if client config folders do not follow these naming conventions!

Client Folder Naming Rules

All client configuration folders under config/clients/ MUST follow these strict naming conventions:

| ❌ INVALID | ✅ VALID | Reason | | ------------------- | ------------------- | ---------------------------------- | | LaSalle/ | lasalle/ | Must be lowercase | | LaSalleAM/ | lasalle_am/ | Must be lowercase with underscores | | Black Stone/ | blackstone/ | No spaces allowed | | ABG-Client/ | abg_client/ | Use underscores, not hyphens | | Starwood Capital/ | starwood_capital/ | No spaces, all lowercase |

Recent Client Folder Renames

The following client folders have been renamed to comply with these requirements:

• LaSalle/ → lasalle/
• LaSalleAM/ → lasalle_am/

Requirements Summary

✅ MUST use:

• All lowercase letters
• Underscores (_) for word separation

❌ MUST NOT use:

• Capital letters
• Spaces
• Hyphens (-)
• Special characters

Why This Matters

The deployment scripts and runtime operations rely on these folder names for:

• Kubernetes resource naming
• Cloud Workflow identifiers
• File path resolution
• CI/CD pipeline operations

Failure to follow these conventions will result in deployment failures and runtime errors!

Features

• Flow Output References: Reference outputs from previous flows using {{flowName.output}}, {{flowName.rentrolls}}, {{flowName.rentroll_units}}
• API-Specific Output Types: API flows generate separate rentrolls and rentroll_units outputs
• Template Variable System: Use template variables in YAML configuration files for dynamic values
• Flow Validation: Compile-time validation of flow references and output types
• Multiple Data Sources: Support for API, CSV, and other data source types
• Data Processing: SQL and JavaScript-based data transformation
• Flow-based Configuration: Define complex data processing workflows in YAML
• CLI Interface: Command-line interface for running flows with parameters

Template Variables & Flow References

The system supports two types of template variables in YAML configuration files:

1. Basic Template Variables

Use {{variable}} syntax for CLI parameters and simple values:

• {{outputFolder}} - Output directory path
• {{dateFrom}} - Start date for API extraction
• {{dateTo}} - End date for API extraction
• {{databaseIds}} - Database IDs (comma-separated or "all")

2. Flow Output References

Use {{flowName.outputType}} syntax to reference outputs from previous flows:

API Flow Outputs

API flows generate two separate data outputs:

• {{flowName.rentrolls}} - Rentrolls CSV file
• {{flowName.rentroll_units}} - Rentroll units CSV file

⚠️ Database Restrictions:

• Single Database: Direct references work (e.g., databaseIds: "123")
• Multiple Databases: Direct references are ambiguous and will fail (e.g., databaseIds: "all" or "123,124,125")

For multiple databases, you must add a process block to combine the data, then reference .output.

Processed Flow Outputs

Flows with process blocks generate:

• {{flowName.output}} - Processed output file (CSV or Parquet)

Output Reference Rules

The system enforces strict validation rules for flow output references:

✅ Valid References

# API extract-only flow
- name: api_extract
  sources:
    - id: units
      type: api
      config:
        databaseIds: '{{databaseIds}}'
        environment: staging

# Valid usage of API outputs
- name: process_units
  sources:
    - id: raw_units
      type: csv
      config:
        path: '{{api_extract.rentroll_units}}' # ✅ API flow, valid output
  process:
    type: sql
    content: 'SELECT * FROM raw_units'

- name: final_report
  sources:
    - id: processed_data
      type: csv
      config:
        path: '{{process_units.output}}' # ✅ Flow with process block

❌ Invalid References

# This will cause a validation error:
- name: invalid_reference
  sources:
    - id: data
      type: csv
      config:
        path: '{{api_extract.output}}' # ❌ API without process can't use .output

# This will cause an ambiguity error:
- name: api_extract_multiple
  sources:
    - id: units
      type: api
      config:
        databaseIds: 'all' # or "123,124,125"
        environment: staging

- name: ambiguous_reference
  sources:
    - id: data
      type: csv
      config:
        path: '{{api_extract_multiple.rentrolls}}' # ❌ Multiple databases are ambiguous

Error Message:

Cannot reference {{api_extract_multiple.rentrolls}} because flow 'api_extract_multiple'
extracts from multiple databases. The paths for .rentrolls and .rentroll_units are
ambiguous when multiple databases are used. Consider adding a process block to combine
the data, then reference {{api_extract_multiple.output}} instead.

Complete Example

Single Database Scenario (Direct References)

flows:
  # Step 1: Extract data from API (single database)
  - name: api_get_data
    sources:
      - id: units
        type: api
        config:
          dateFrom: '{{dateFrom}}'
          dateTo: '{{dateTo}}'
          databaseIds: '123' # Single database - direct references work
          environment: staging

  # Step 2: Process units data
  - name: clean_units
    sources:
      - id: raw_units
        type: csv
        config:
          path: '{{api_get_data.rentroll_units}}' # ✅ Works for single database
    process:
      type: javascript
      file: src/clients/LaSalle/las-units-clean.ts

  # Step 3: Process rentrolls data
  - name: clean_rentrolls
    sources:
      - id: raw_rentrolls
        type: csv
        config:
          path: '{{api_get_data.rentrolls}}' # ✅ Works for single database
    process:
      type: javascript
      file: src/clients/LaSalle/las-rentrolls-clean.ts

  # Step 4: Join cleaned data
  - name: final_report
    sources:
      - id: units_data
        type: csv
        config:
          path: '{{clean_units.output}}'
      - id: rentrolls_data
        type: csv
        config:
          path: '{{clean_rentrolls.output}}'
    process:
      type: sql
      content: |
        SELECT u.*, r.rent_amount
        FROM units_data u
        LEFT JOIN rentrolls_data r ON u.unit_id = r.unit_id
        AND u._inprocess_databaseId = r._inprocess_databaseId

Multiple Database Scenario (Process Block Required)

flows:
  # Step 1: Extract data from API (multiple databases)
  - name: api_get_data
    sources:
      - id: units
        type: api
        config:
          dateFrom: '{{dateFrom}}'
          dateTo: '{{dateTo}}'
          databaseIds: 'all' # Multiple databases - must use process block
          environment: staging
    process:
      type: sql
      content: |
        SELECT *, _inprocess_databaseId 
        FROM units

  # Step 2: Use combined output (not individual database files)
  - name: clean_units
    sources:
      - id: combined_units
        type: csv
        config:
          path: '{{api_get_data.output}}' # ✅ Works - references processed combined data
    process:
      type: javascript
      file: src/clients/LaSalle/las-units-clean.ts

CLI Usage

Running Flows

Flows are transformed at build time and copied to dist/config. The flow script builds first, so always point --file/-f to the dist path when running flows.

# Example (production, single DB)
pnpm run flow -- -f dist/config/clients/lasalle/production/main-flow.yaml \
  --dateFrom "2025-05-01" \
  --dateTo "2025-08-31" \
  --databaseIds "434"

# Example (staging, multiple DBs)
pnpm run flow -- -f dist/config/clients/lasalle/staging/main-flow.yaml \
  --outputFolder output/all_dbs \
  --databaseIds "all" \
  --dateFrom "2025-01-01" \
  --dateTo "2025-01-31"

Note: The flow script also accepts config/... paths and will rewrite them to dist/config/... automatically. Example:

pnpm run flow -- -f config/clients/lasalle/production/main-flow.yaml \
  --dateFrom "2025-05-01" --dateTo "2025-08-31" --databaseIds "434"

Using DuckLake API

DuckLake can be enabled to route queries to the DuckLake-backed internal endpoint.

• Set the environment variable USE_DUCKLAKE_API to true or 1.
• When enabled, the query path switches from /thirdparty/query to /proda-internal/ducklake/query against the same apiBasePath.
• No other configuration changes are required.

Examples:

# Enable DuckLake for a single run
USE_DUCKLAKE_API=1 pnpm run flow -- -f dist/config/clients/lasalle/staging/main-flow.yaml \
  --outputFolder output/test \
  --databaseIds "123" \
  --dateFrom "2025-08-01" \
  --dateTo "2025-08-31"

# Or export in your shell session
export USE_DUCKLAKE_API=true
pnpm run flow -- -f dist/config/clients/lasalle/production/main-flow.yaml \
  --databaseIds "434" \
  --dateFrom "2025-05-01" \
  --dateTo "2025-08-31"

Notes:

• This setting affects API extraction flows only.
• Ensure your apiBasePath (from config/environments.yaml) is reachable for the chosen environment. Internal paths may require running inside the cluster.

Validation uses the source config/... path and does not rewrite:

Validating Flows (Dry Run)

The validation command allows you to test your flow configurations without executing them. It outputs the fully resolved YAML configuration with all template variables substituted:

# Validate flow configuration and show resolved YAML
pnpm run validate -- \
  --file config/clients/lasalle/staging/main-flow.yaml \
  --outputFolder output/test \
  --databaseIds "123"

# Alternative syntax
pnpm run flow -- --validate \
  --file config/clients/lasalle/staging/main-flow.yaml \
  --outputFolder output/test \
  --databaseIds "434"

What Validation Does:

• ✅ Template Resolution: Substitutes all {{variable}} and {{flow.output}} references
• ✅ Reference Validation: Verifies all flow output references are valid
• ✅ Configuration Expansion: Expands API credentials and environment settings
• ✅ Error Detection: Catches invalid references before execution

Example Output:

$schema: src/infra/adapters/config/flow-schema.yaml
flows:
  - name: api_extract
    sources:
      - id: units
        type: api
        config:
          dateFrom: 2025-07-29T00:00:00.000Z
          dateTo: 2025-08-28T00:00:00.000Z
          databaseIds: ['123']
          environment:
            apiBasePath: https://app.proda.ai
          outputFolder: ./output/
    process:
      type: sql
      content: |
        SELECT *, _inprocess_databaseId 
        FROM units
  - name: clean_data
    sources:
      - id: raw_data
        type: csv
        config:
          path: output/test/api_extract_processed.csv # ← Template resolved!
    process:
      type: sql
      content: SELECT unit_id, building_name FROM raw_data

This shows exactly how your configuration will be processed, making it easy to spot issues with template variables or flow references.

The CLI arguments are automatically converted to template parameters and validated:

• --outputFolder becomes {{outputFolder}}
• --databaseIds becomes {{databaseIds}}
• --dateFrom becomes {{dateFrom}}
• --dateTo becomes {{dateTo}}

Flow Validation

The system validates all flow references at load time:

• Missing flows: Flow 'nonexistent_flow' not found
• Invalid output types: Output type 'output' is not valid for flow 'api_extract'
• Forward references: Flows can only reference previous flows
• Type safety: API flows can't reference non-existent output types
• Database ambiguity: Direct references to .rentrolls/.rentroll_units fail for multiple databases

Error Handling

The system provides comprehensive error handling for template variables and flow references:

• Missing template variables: Hard failure with list of available variables
• Invalid flow references: Hard failure with list of available flows
• Invalid output types: Hard failure with list of valid outputs for that flow type
• Forward references: Hard failure when referencing flows that come later
• Database ambiguity: Clear error message with guidance to use process blocks
• File not found: Clear error message about missing flow output files

Database ID Processing

The system handles different database ID formats:

• Single ID: --databaseIds "116" → API extracts one database, adds _inprocess_databaseId: "116"
• Multiple IDs: --databaseIds "116,117,118" → API extracts all specified databases with their IDs
• All databases: --databaseIds "all" → API extracts all available databases with their IDs

The _inprocess_databaseId column is automatically added to all extracted data, enabling database-specific processing and joins.

Best Practices

1. Use Specific Output References for API Flows

# Good: Specific to data type
path: "{{api_extract.rentroll_units}}"  # For unit processing
path: "{{api_extract.rentrolls}}"       # For rent processing

# Avoid: Generic reference when you need specific data
path: "{{api_extract.output}}"          # Only if flow has process block

2. Organize Flows Logically

# Good flow organization
flows:
  - name: api_get_data # 1. Extract data
  - name: clean_units # 2. Process units
  - name: clean_rentrolls # 3. Process rentrolls
  - name: join_data # 4. Combine results

3. Leverage the _inprocess_databaseId Column

SELECT
  _inprocess_databaseId,
  COUNT(*) as units_per_database
FROM units
GROUP BY _inprocess_databaseId

4. Add Process Blocks for Combined Data Access

- name: api_with_processing
  sources:
    - id: units
      type: api
      config:
        databaseIds: '{{databaseIds}}'
  process:
    type: sql
    content: 'SELECT *, _inprocess_databaseId FROM units' # Now supports .output

Advantages of This Approach

✅ Type Safety: Prevents invalid output type references at load time
✅ Clear Separation: Distinguishes between raw API data and processed outputs
✅ Explicit Dependencies: Flow dependencies are clear from the YAML structure
✅ Better Error Messages: Failed references provide clear context and suggestions
✅ Flexible Data Flow: Can use both raw and processed data as needed
✅ Validation at Parse Time: Catches errors before execution begins

System Architecture

The Bulk Output API is a cloud-native data processing system that extracts, transforms, and loads data from the Proda API for various clients. The system uses Google Cloud infrastructure with Kubernetes for scalable processing and supports multiple output formats.

Clean Architecture

The system follows clean architecture principles with clear separation of concerns:

Domain Layer (src/domain/)

• Flow Models: Core business entities and types
• Flow Validation: Pure business logic for validating flow configurations
• Template Processing: Core template substitution and processing logic

Infrastructure Layer (src/infra/adapters/)

• YAML Parsing: File system operations and YAML parsing
• CLI Integration: Command-line argument processing
• API Clients: External API communication
• Data Processing: Polars-based data transformation

graph TB
    subgraph "External Systems"
        API["Proda API"]
        Auth["Auth0"]
        Webhook["Client Webhooks"]
    end

    subgraph "Configuration"
        MainFlow["main-flow.yaml"]
        CloudFlow["cloud-workflow.yaml"]
        EnvConfig["environments.yaml"]
    end

    subgraph "Google Cloud Infrastructure"
        Scheduler["Cloud Scheduler"]
        Workflow["Cloud Workflows"]

        subgraph "Kubernetes Cluster"
            Job["Kubernetes Jobs"]
            Pod["Processing Pods"]
        end

        GCS["Cloud Storage"]
        SecretManager["Secret Manager"]
    end

    subgraph "Data Processing Pipeline"
        Extract["Data Extraction"]
        Transform["Data Transformation"]
        Load["Data Loading"]
        Archive["Archive & Upload"]
    end

    subgraph "Monitoring & Telemetry"
        Prometheus["Prometheus"]
        Datadog["Datadog"]
        Pushgateway["Pushgateway"]
    end

    subgraph "Client-Specific Processing"
        ABG["ABG Client"]
        BX["Blackstone"]
        LaSalle["LaSalle Client"]
        Starwood["Starwood Client"]
    end

    %% Flow connections
    Scheduler --> Workflow
    Workflow --> Job
    Job --> Pod

    Pod --> Extract
    Extract --> Transform
    Transform --> Load
    Load --> Archive
    Archive --> GCS

    %% Configuration connections
    MainFlow --> Pod
    CloudFlow --> Workflow
    EnvConfig --> Pod

    %% External system connections
    API --> Extract
    Auth --> Extract
    Archive --> Webhook

    %% Security connections
    SecretManager --> Pod

    %% Client processing
    Transform --> ABG
    Transform --> BX
    Transform --> LaSalle
    Transform --> Starwood

    %% Monitoring connections
    Pod --> Pushgateway
    Pod --> Datadog
    Pushgateway --> Prometheus

    %% Styling
    classDef external fill:#e1f5fe
    classDef config fill:#f3e5f5
    classDef cloud fill:#e8f5e8
    classDef processing fill:#fff3e0
    classDef monitoring fill:#fce4ec
    classDef client fill:#f0f4c3

    class API,Auth,Webhook external
    class MainFlow,CloudFlow,EnvConfig config
    class Scheduler,Workflow,Job,Pod,GCS,SecretManager cloud
    class Extract,Transform,Load,Archive processing
    class Prometheus,Datadog,Pushgateway monitoring
    class ABG,BX,LaSalle,Starwood client

Key Components

• Data Sources: Extracts data from Proda API with authentication via Auth0
• Configuration-Driven: Uses YAML files to define client-specific workflows and processing logic
• Flow Validation: Compile-time validation ensures flow references are correct before execution
• Template Processing: Dynamic configuration with type-safe template variable substitution
• Scalable Processing: Kubernetes jobs with nodejs-polars for efficient data transformation
• Multi-Format Output: Supports CSV and Parquet formats stored in Google Cloud Storage
• Monitoring: Dual telemetry with Prometheus and Datadog for comprehensive observability
• Client-Specific Logic: Customizable processing pipelines for different business requirements

Table of Contents

1. Folder Structure
2. Exporting a Database Using the API
3. Creating a New Workflow Configuration
4. Deploying the Workflow
5. Example Configuration
6. Adding Telemetry
7. Data Processing with nodejs-polars
8. Metrics and Monitoring
9. Using DuckLake API

Running locally

The system is configured to run on the proda cluster.

1. open config/environments.yaml
2. uncomment line 2 and 7
3. comment line 3 and 8

End result for local execution:

staging:
  apiBasePath: 'https://staging.proda.ai'
  # apiBasePath: "http://excelsior-api-service.excelsior.svc.cluster.local"
  apiAuthUrl: 'https://proda-development.eu.auth0.com/oauth/token'
  apiAudience: 'https://staging.proda.ai/thirdparty/'
production:
  apiBasePath: 'https://app.proda.ai'
  # apiBasePath: "http://excelsior-api-service.excelsior.svc.cluster.local"
  apiAuthUrl: 'https://proda.eu.auth0.com/oauth/token'
  apiAudience: 'https://app.proda.ai/thirdparty/'

Save and try again

Running locally with dbt as source

You can run an export locally against a dbt-backed query service (e.g. http://localhost:3125) while authenticating against the staging environment:

pnpm run export \
    --dateFrom 2025-01-01 --dateTo 2026-01-31 \
    --queryServiceApiBasePath http://localhost:3125 \
    --apiBasePath https://staging.proda.ai \
    --apiAudience https://staging.proda.ai/thirdparty/ \
    --apiAuthUrl https://proda-development.eu.auth0.com/oauth/token

This requires the query-service to be reachable at the --queryServiceApiBasePath address. Either run the query-service locally, or port-forward it from Kubernetes, for example:

kubectl port-forward -n query-builder svc/query-service 3125:80

Regenerating OpenAPI Client Libraries

To regenerate the OpenAPI client libraries, you can use the provided script in the package.json file. This script will generate the client types based on the current OpenAPI specification stored in proda-openapi.json

Steps to Regenerate

1. Ensure you have all the necessary dependencies installed by running:

pnpm install

1. Run the script to generate the client libraries:

pnpm run api:gen-client

Folder Structure

The project's configuration files are organized in the following structure:

  config/
    clients/
      <client_name>/
        cloud-workflow.yaml
        <environment>/
          main-flow.yaml

• clients: This directory contains folders for each client. For example, bx and vts.
• client_name: Replace client_name with the actual client identifier, such as bx or vts.
• environment: This represents the environment such as staging or production.
• main-flow.yaml: The main workflow configuration file specific to the environment. Here are the details of the export we want to do.
• cloud-workflow.yaml: The google workflow configuration file to deploy the workflow.

Exporting a Database Using the API

1. checkout the code or use codespace
2. open a terminal window (top menu -> Terminal -> New Terminal)
3. Type pnpm install to setup the project and get all the libraries needed
4. Get an API key from docs.proda.ai or from the staging environment and copy it
5. In the same terminal type export PRODA_API_KEY=<API KEY>
6. Run the flow with the following command

ts-node src/clients/proda/api-export.ts -d <databaseId> -e <staging|production> -f <fromDate> -t <toDate>

You can also use the pnpm script:

pnpm run export -- -f <fromDate> -t <toDate> -d <databaseId> -e <environment>

Available Options

Options:
  --version                   Show version number                      [boolean]
  -f, --dateFrom              Start date for the export (YYYY-MM-DD)   [string] [required]
  -t, --dateTo                End date for the export (YYYY-MM-DD)     [string] [default: current date + 1 year]
  -d, --databaseIds           Comma-separated list of database IDs or "all" [string] [default: "all"]
  -c, --columns               Comma-separated list of columns or "all" [string] [default: "all"]
  -i, --includeDeprecatedColumns Include deprecated columns (valid only when columns="all") [boolean] [default: false]
  -e, --environment           The environment to use                   [string]
  --apiBasePath, --ab         The base path for the API                [string]
  --apiAuthUrl, --au          The Auth0 URL to get an access token. This is env specific [string]
  --apiAudience, --ad         The audience that will be allowed to use the token. This is env specific [string]
  -o, --outputFolder          The folder to save the exported data     [string] [default: "./output/"]
  -h, --help                  Show help                                [boolean]

Creating a New Workflow Configuration

To create a new workflow for a client in a specific environment, follow these steps:

Step 1: Create the Directory Structure

Navigate to the config/clients directory

cd path/to/project/config/clients

Create a directory for the client if it doesn't exist

mkdir -p client_name/environment

Replace client_name with the client's identifier (e.g., bx) and environment with staging or production.

Step 2: Create the flow Configuration Files

Create main-flow.yaml for each environment

$schema: 'src/infra/adapters/config/flow-schema.yaml'
flows:
  - name: api_get_data
    sources:
      - id: units
        type: api
        config:
          dateFrom: '{{dateFrom}}'
          dateTo: '{{dateTo}}'
          databaseIds: '{{databaseIds}}'
          environment: staging

  - name: process_data
    sources:
      - id: raw_units
        type: csv
        config:
          path: '{{api_get_data.rentroll_units}}'
    process:
      type: sql
      content: |
        SELECT *, 
               _inprocess_databaseId,
               COUNT(*) OVER () as total_records
        FROM raw_units
        WHERE unit_id IS NOT NULL

Flow format

You can use TypeScript or SQL for the data processing. For more details on the main-flow.yaml format, see the documentation in docs/main-flow.yaml.md.

For SQL functions supported in data processing, check the Polars SQL functions reference.

Flow Output Types

Each flow generates specific output types based on its configuration:

• API flows without process: Only .rentrolls and .rentroll_units
• API flows with process: .output, .rentrolls, and .rentroll_units
• Non-API flows with process: Only .output

Create or Update cloud-workflow.yaml

This file serves as a base template for your workflow, containing shared logic and configurations. You can modify it to fit specific needs or use it as a starting point for the main-flow.yaml configurations.

Deploying the Workflow

Prerequisites

Ensure you have the Google Cloud SDK (gcloud) installed and configured with the necessary permissions. Install Node.js and ts-node to run the TypeScript script.

pnpm install -g ts-node

Deploy the Workflow

Run the deployment script with the appropriate arguments:

cd infra/deploy
ts-node deploy.ts -e environment -d dockerImageTag

Replace environment with either staging or production and dockerImageTag with the appropriate Docker image version tag.

The script will automatically:

1. Detect the client directories and corresponding main-flow.yaml files.
2. Deploy the Google Cloud Workflow using the main-flow.yaml configuration.
3. Manage Cloud Scheduler jobs for the deployed workflows.

Example Configuration

Here's an example configuration for a client named bx:

Directory Structure:

config/
  clients/
    bx/
      staging/
        main-flow.yaml
      production/
        main-flow.yaml
      cloud-workflow.yaml

Adding Telemetry

Telemetry Wrapper

There is a wrapper function to send telemetry data to Datadog. This function will log the start time, end time, and time taken for each API call. Additionally, it will log any errors or failures.

To enable

provide the env variable DATADOG_API_KEY. If the variable it not provided, the output will go to stdout

Data Processing with nodejs-polars

This project uses nodejs-polars for efficient data processing and manipulation. Polars is a fast DataFrame library implemented in Rust with bindings for Node.js.

Key Features

• High-performance DataFrame operations
• SQL query support for data transformation
• Series manipulation with various operations
• Support for both ESM and CommonJS imports

Basic Usage in Flows

In your flow files, you can use Polars for data manipulation either through TypeScript code or SQL queries:

// In TypeScript processing files
import pl from 'nodejs-polars';

// Create a DataFrame from a source
const df = pl.readCSV('path/to/file.csv');

// Perform operations
const processedDf = df.filter(pl.col('column_name').gt(0)).select(['column1', 'column2']).sort('column1');

// Export the result
processedDf.writeCSV('output.csv');

For SQL processing in main-flow.yaml, the system uses Polars' SQL capabilities to execute queries on the data sources.

Troubleshooting Function Issues

When debugging data processing functions, you can add temporary logging and data collection to help identify issues. Here's an example of how to add debugging code to track aggregation results:

// Example: Debugging aggregation results for security deposits
const sumSecurityDeposit = filteredUnits
  .groupBy(groupByColumns)
  .agg(col('security_deposit_amount').cast(pl.Float64).sum().alias('sum_security_deposit_amount'));

logger.info('SUM Security Deposit');
const sumSecurityDepositResult = await sumSecurityDeposit.filter(col('external_lease_id').eq(lit('169545'))).collect();
sumSecurityDepositResult
  .toRecords()
  .forEach((record) => logger.info(`${record.external_lease_id} - ${record.sum_security_deposit_amount}`));

This pattern allows you to:

• Create intermediate aggregations for debugging
• Filter specific records to inspect data
• Log the results to track data transformations
• Verify calculations before proceeding with the main processing

You can apply similar debugging techniques throughout your data processing functions to isolate and identify issues.

Metrics and Monitoring

This application supports dual telemetry with both Datadog and Prometheus metrics:

Prometheus Metrics

The application uses Prometheus Pushgateway for job-based metrics collection:

• Function Duration: bulk_output_function_duration_seconds - Histogram of function execution times
• Function Calls: bulk_output_function_calls_total - Counter of total function calls
• Function Errors: bulk_output_function_errors_total - Counter of function errors by type
• Function Status: bulk_output_function_status - Gauge showing current function status

Environment Variables

• PUSHGATEWAY_URL: URL of the Prometheus Pushgateway (automatically set by Terraform)
• DATADOG_API_KEY: Datadog API key for legacy metrics
• ENVIRONMENT: Environment name (staging/production)

Usage Example

import { createTelemetryAdapter } from './src/infra/adapters/telemetry';

const telemetry = createTelemetryAdapter({
  pushgatewayUrl: process.env.PUSHGATEWAY_URL,
  datadogApiKey: process.env.DATADOG_API_KEY,
  environment: process.env.ENVIRONMENT,
  jobName: 'bulk-output-api',
});

// Use with any async function
const result = await telemetry(
  async () => {
    // Your function logic here
    return await processData();
  },
  'processData',
  { client: 'bx', workflow: 'export' }
);

Infrastructure

The Prometheus Pushgateway is automatically deployed via Terraform:

• Deployed in the prometheus namespace (shared monitoring infrastructure)
• Accessible at prometheus-pushgateway.prometheus.svc.cluster.local:9091
• Automatically discovered by kube-prometheus-stack via ServiceMonitor
• Includes persistent storage for metrics between restarts
• Network policies allow access from application namespaces

Conclusion

This system efficiently extracts and exports data from APIs to various output formats using a robust, type-safe flow configuration system. With flow output references, template variables, and comprehensive validation, it provides a flexible and maintainable way to build complex data processing pipelines.