cursus.steps.scripts.dummy_data_loading

Dummy Data Loading Processing Script

This script processes user-provided data instead of calling internal Cradle services. It serves as a drop-in replacement for CradleDataLoadingStep by reading data from an input channel, generating schema signatures and metadata, and outputting the processed data in the same format as the original Cradle data loading step.

ensure_directory(directory)[source]

Ensure a directory exists, creating it if necessary.

optimize_dtypes(df, log_func=None)[source]

Optimize DataFrame dtypes to reduce memory usage.

Applies the following optimizations: - Downcast numeric types (int64->int32, float64->float32) - Convert object columns with low cardinality to category

Parameters:

  • df (DataFrame) – Input DataFrame

  • log_func (Callable | None) – Optional logging function

Returns:

DataFrame with optimized dtypes

Return type:

DataFrame

detect_file_format(file_path)[source]

Detect the format of a data file based on extension and content.

Parameters:

file_path (Path) – Path to the data file

Returns:

‘csv’, ‘parquet’, ‘json’, or ‘unknown’

Return type:

String indicating the format

read_data_file(file_path, file_format)[source]

Read a data file based on its format.

Parameters:

  • file_path (Path) – Path to the data file

  • file_format (str) – Format of the file (‘csv’, ‘parquet’, ‘json’)

Returns:

DataFrame containing the data

Raises:
Return type:

DataFrame

generate_schema_signature(df)[source]

Generate a schema signature from a DataFrame.

The schema signature is just a list of column names from the input data.

Parameters:

df (DataFrame) – DataFrame to analyze

Returns:

List of column names

Return type:

List[str]

generate_metadata(df)[source]

Generate metadata information from a DataFrame.

Parameters:

df (DataFrame) – DataFrame to analyze

Returns:

Dictionary containing metadata information

Return type:

Dict[str, Any]

generate_mods_metadata(df)[source]

Generate MODS-compatible CSV metadata from a DataFrame.

MODS metadata format is a simple CSV with 3 columns: - varname: Column name - iscategory: “true” if string/object/category type, “false” otherwise - datatype: pandas dtype as string

This lightweight format can be generated from the first batch only, enabling true streaming mode without needing the full DataFrame.

Parameters:

df (DataFrame) – DataFrame to analyze (typically first batch)

Returns:

List of lists representing CSV rows [header, row1, row2, …]

Return type:

List[List[str]]

find_data_files(input_dir)[source]

Find all data files in the input directory.

Parameters:

input_dir (Path) – Directory to search for data files

Returns:

List of paths to data files

Return type:

List[Path]

combine_files(data_files, max_workers=None, batch_size=10, streaming_batch_size=None)[source]

Combine multiple data files using parallel processing and optional streaming.

Uses parallel file reading and batch concatenation for improved performance. Memory-efficient approach with optional streaming mode.

Streaming Mode: When streaming_batch_size is set, processes files in batches to avoid loading all DataFrames into memory simultaneously. This is the most memory-efficient mode.

Parameters:

  • data_files (List[Path]) – List of data file paths

  • max_workers (int | None) – Maximum number of parallel workers (default: cpu_count)

  • batch_size (int) – Number of DataFrames to concatenate at once (default: 10)

  • streaming_batch_size (int | None) – Number of files to process per batch (enables streaming mode) - If None: Loads all files into memory (original behavior) - If set: Processes files in batches, concatenating incrementally - Recommended: 10-20 files per batch for memory-constrained environments

Returns:

Combined DataFrame from all files

Return type:

DataFrame

process_data_files(data_files)[source]

DEPRECATED: Legacy function for backward compatibility. Use combine_files() instead for better performance and memory efficiency.

Process multiple data files and combine them into a single DataFrame.

Parameters:

data_files (List[Path]) – List of data file paths

Returns:

Combined DataFrame

Return type:

DataFrame

write_signature_file(signature, output_dir)[source]

Write the signature file to the output directory in CSV format.

The signature file contains column names separated by commas, matching the format expected by tabular_preprocessing script.

Parameters:

  • signature (List[str]) – Schema signature list of column names

  • output_dir (Path) – Output directory path

Returns:

Path to the written signature file

Return type:

Path

write_metadata_file(metadata, output_dir, format='JSON')[source]

Write the metadata file to the output directory in specified format.

Supports two formats: - JSON: Detailed metadata with statistics (requires Dict input) - MODS: Simple CSV with 3 columns (requires List[List[str]] input)

Parameters:

  • metadata (Dict[str, Any] | List[List[str]]) – Metadata as Dict (JSON) or List[List[str]] (MODS CSV)

  • output_dir (Path) – Output directory path

  • format (str) – Output format - “JSON” or “MODS” (default: “JSON”)

Returns:

Path to the written metadata file

Raises:

ValueError – If format is unsupported or metadata type doesn’t match format

Return type:

Path

write_single_shard(df, output_dir, shard_index, output_format)[source]

Write a single data shard in the specified format.

Parameters:

  • df (DataFrame) – DataFrame to write

  • output_dir (Path) – Output directory path

  • shard_index (int) – Index of the shard (for filename)

  • output_format (str) – Output format (‘CSV’, ‘JSON’, ‘PARQUET’)

Returns:

Path to the written shard file

Raises:
Return type:

Path

write_data_shards(df, output_dir, shard_size, output_format)[source]

Write DataFrame as multiple data shards.

Parameters:

  • df (DataFrame) – DataFrame to write

  • output_dir (Path) – Output directory path

  • shard_size (int) – Number of rows per shard

  • output_format (str) – Output format (‘CSV’, ‘JSON’, ‘PARQUET’)

Returns:

List of paths to written shard files

Return type:

List[Path]

write_single_data_file(df, output_dir, output_format)[source]

Write DataFrame as a single data file.

Parameters:

  • df (DataFrame) – DataFrame to write

  • output_dir (Path) – Output directory path

  • output_format (str) – Output format (‘CSV’, ‘JSON’, ‘PARQUET’)

Returns:

Path to the written data file

Raises:
Return type:

Path

write_data_output(df, output_dir, write_shards=False, shard_size=10000, output_format='CSV')[source]

Write data output - either as shards or single file based on configuration.

Parameters:

  • df (DataFrame) – Processed DataFrame

  • output_dir (Path) – Output directory path

  • write_shards (bool) – If True, write data as shards; if False, write single file

  • shard_size (int) – Number of rows per shard file

  • output_format (str) – Output format (‘CSV’, ‘JSON’, ‘PARQUET’)

Returns:

Path to single data file or list of shard file paths

Return type:

Path | List[Path]

process_first_batch_for_metadata(first_batch_files, metadata_format, max_workers, batch_size)[source]

Process first batch of files to generate signature and metadata.

Extracts signature and metadata from the first batch only, enabling streaming mode to proceed without loading the full dataset.

Parameters:

  • first_batch_files (List[Path]) – List of file paths in first batch

  • metadata_format (str) – “JSON” or “MODS”

  • max_workers (int | None) – Number of parallel workers

  • batch_size (int) – DataFrame concat batch size

Returns:

Tuple of (signature, metadata, first_batch_df)

Return type:

tuple

write_batch_as_shards(df, output_dir, shard_counter, shard_size, output_format)[source]

Write DataFrame batch as shards with continuous numbering.

Parameters:

  • df (DataFrame) – DataFrame to write as shards

  • output_dir (Path) – Output directory path

  • shard_counter (int) – Starting shard index number

  • shard_size (int) – Rows per shard

  • output_format (str) – Output data format

Returns:

Tuple of (written_shard_paths, updated_counter)

Return type:

tuple

process_remaining_batches(remaining_files, data_output_dir, shard_counter, streaming_batch_size, shard_size, output_format, max_workers, batch_size)[source]

Stream and write remaining file batches.

Processes remaining files in batches, writing shards incrementally without loading the full dataset into memory.

Parameters:

  • remaining_files (List[Path]) – List of remaining file paths to process

  • data_output_dir (Path) – Output directory for data shards

  • shard_counter (int) – Starting shard index number

  • streaming_batch_size (int) – Number of files per batch

  • shard_size (int) – Rows per shard

  • output_format (str) – Output data format

  • max_workers (int | None) – Number of parallel workers

  • batch_size (int) – DataFrame concat batch size

Returns:

Tuple of (all_written_shards, total_rows_processed, final_counter)

Return type:

tuple

process_streaming_mode(data_files, signature_output_dir, metadata_output_dir, data_output_dir, metadata_format, streaming_batch_size, shard_size, output_format, max_workers, batch_size)[source]

True streaming mode: Never loads full DataFrame into memory.

Process data files in batches, generating outputs incrementally: 1. First batch → signature & metadata (from first batch only) 2. All batches → write shards incrementally 3. Free memory after each batch

Memory usage: ~1-2GB per batch (not dependent on total data size) Scales to: ANY data size (10GB, 100GB, 1TB+)

Parameters:

  • data_files (List[Path]) – List of data file paths

  • signature_output_dir (Path) – Directory for signature output

  • metadata_output_dir (Path) – Directory for metadata output

  • data_output_dir (Path) – Directory for data shard output

  • metadata_format (str) – “JSON” or “MODS”

  • streaming_batch_size (int) – Number of files per batch

  • shard_size (int) – Rows per output shard

  • output_format (str) – Output data format

  • max_workers (int | None) – Number of parallel workers

  • batch_size (int) – DataFrame concat batch size

Returns:

Dictionary of output file paths

Return type:

Dict[str, Path | List[Path]]

main(input_paths, output_paths, environ_vars, job_args=None)[source]

Main entry point for the Dummy Data Loading script.

Parameters:

  • input_paths (Dict[str, str]) – Dictionary of input paths with logical names

  • output_paths (Dict[str, str]) – Dictionary of output paths with logical names

  • environ_vars (Dict[str, str]) – Dictionary of environment variables

  • job_args (Namespace | None) – Command line arguments (optional)

Returns:

Dictionary of output file paths

Return type:

Dict[str, Path | List[Path]]