Troubleshoot Pipeline Performance and Memory Usage

Concepts

This topic requires an understanding of pipeline batches, which are explained in The Lifecycle of a Pipeline.

Detect Slow Performance and High Memory Usage

Run PROFILE PIPELINE

Run PROFILE PIPELINE to gather resource consumption metrics, such as starting and ending times, for operations that a batch processes. Partition specific metrics are not included; to retrieve those query the information schema view specified in the next section.

Query the information_schema.PIPELINE_BATCHES_SUMMARY View

The following query returns diagnostics for each batch associated with a pipeline that writes to a database.

SELECT DATABASE_NAME, PIPELINE_NAME, BATCH_ID, BATCH_STATE, BATCH_TIME
FROM information_schema.PIPELINES_BATCHES_SUMMARY
ORDER BY BATCH_ID;
  • BATCH_STATE: If the BATCH_STATE of a batch is Queued, it indicates that the batch cannot extract, shape, or load data until cluster resources are freed.

  • A high BATCH_TIME value indicates a long amount of time for the batch to extract, shape, and to load the data (combined).

For complete reference and description of all the fields, refer to information_schema.PIPELINE_BATCHES_SUMMARY.

Query the information_schema.MV_ACTIVITIES and information_schema.MV_QUERIES Views

As an alternative to querying the information_schema.PIPELINE_BATCHES_SUMMARY table, as detailed in the previous section, you can run the following query, which provides diagnostic information per pipeline, such as avg_cpu_time_ms, avg_network_time_ms, avg_disk_time_ms, and avg_memory_mb.

SELECT substr(replace(replace(MVAC.activity_name,'\n',''), ' ',' '),1,30) AS ACTIVITY_n,
substr(replace(replace(query_text,'\n',''), ' ',' '),1,30) AS query_text,
database_name AS databaseName,last_finished_timestamp AS last_run,
(cpu_time_ms/(run_count+success_count+failure_count)) AS avg_cpu_time_ms,
(cpu_wait_time_ms/(run_count+success_count+failure_count)) AS avg_cpu_wait_time_ms,
(elapsed_time_ms/(run_count+success_count+failure_count)) AS avg_elapsed_time_ms,
(lock_time_ms/(run_count+success_count+failure_count)) AS avg_lock_time_ms,
(network_time_ms/(run_count+success_count+failure_count)) AS avg_network_time_ms,
(disk_time_ms/(run_count+success_count+failure_count)) AS avg_disk_time_ms,
(round((disk_b/1024/1024),2)/(run_count+success_count+failure_count)) AS avg_io_mb,
(round((network_b/1024/1024),2)/(run_count+success_count+failure_count)) AS avg_network_mb,
round((1000*(memory_bs/1024/1024)/(elapsed_time_ms)),2) AS avg_memory_mb,
(memory_major_faults/(run_count+success_count+failure_count)) AS avg_major_faults,
(run_count+success_count+failure_count) AS total_executions
FROM information_schema.mv_activities_cumulative MVAC
JOIN information_schema.mv_queries MVQ ON MVQ.activity_name = MVAC.activity_name
WHERE MVAC.activity_name like '%RunPipeline%' ORDER BY avg_elapsed_time_ms DESC LIMIT 10;

Address Slow Performance and High Memory Usage

Performing the following tasks may improve the pipeline performance and reduce memory usage.

  • Decrease the maximum number of partitions that a batch can utilize using the ALTER PIPELINE SET MAX_PARTITIONS_PER_BATCH command. Decreasing the maximum number of partitions will limit parallelism for large data sources.

  • Increase the batch interval of an existing pipeline using the ALTER PIPELINE SET BATCH_INTERVAL command. If you increase the batch interval, data will be loaded from the data source less frequently.

  • Break large files into smaller chunks to increase the ingestion speed. Large files get parsed on a single leaf, which increases the ingestion time.

  • Add more leaves to your cluster, and then rebalance the cluster, which will result in fewer numbers of partitions per leaf.

  • Are you using a stored procedure with your pipeline? If so, the stored procedure implementation can significantly affect the memory usage and pipeline performance. For details, refer to Writing Efficient Stored Procedures for Pipelines.

  • For low parallelism pipelines, such as single file S3 loads and single partition Kafka topics, you can use an aggregator pipeline by running the CREATE AGGREGATOR PIPELINE command (an aggregator pipeline is not required for single-partition Kafka topics). If you are using an existing non-aggregator pipeline, you cannot convert it directly to an aggregator pipeline. Instead, drop your existing pipeline and recreate it as an aggregator pipeline.

    Caution

    If a pipeline is dropped and recreated, the pipeline will start reading from the earliest offset in the data source, once the pipeline is started. This may cause duplicate records to be inserted into the destination table(s).

  • Have queries or stored procedures that you normally run changed recently? If so, check these queries or stored procedures for changes that could be contributing to slow performance and high memory usage.

  • Do you have any uncommitted transactions? These are transactions where START TRANSACTION is run, but neither COMMIT nor ROLLBACK was run afterwards. For information on resolving issues with uncommitted transactions, refer to Uncommitted Transactions section in ERROR 1205 (HY000): Lock wait timeout exceeded; try restarting transaction.

  • Using the IGNORE or SKIP ALL ERRORS options in the CREATE PIPELINE ... LOAD DATA ... statement may cause the pipeline to run slowly, because these options write large numbers of warnings to the logs. For IGNORE and SKIP ALL ERRORS alternatives, see LOAD DATA.

  • If possible, avoid using the ENCLOSED BY and OPTIONALLY ENCLOSED BY options in your CREATE PIPELINE ... LOAD DATA ... statement.

  • If your pipeline uses a transform, and implements the transform using an interpreted language such as Python, consider using a compiled language to improve the performance.

  • If your pipeline uses a transform, and it is slow, you may be able to use a CREATE PIPELINE ... SET ... WHERE statement instead of using a transform.

  • Check if removing or moving files that were already processed by the pipeline improves pipeline performance. To perform this check, follow these steps:

    1. Run:

      SELECT database_name, pipeline_name, batch_start_unix_timestamp, batch_time - max(batch_partition_time) AS batch_overhead
      FROM information_schema.PIPELINES_BATCHES
      GROUP BY database_name, pipeline_name, batch_id
      ORDER BY database_name, pipeline_name, batch_id;
    2. If batch_overhead is increasing as batch_start_unix_timestamp increases, try the following steps:

      1. Run SELECT * FROM information_schema.PIPELINES_FILES;.

      2. For rows in the output that have the value LOADED in the FILE_STATE column, move or the delete the corresponding files in the FILE_NAME column.

Clear the information_schema.PIPELINE_FILES View

You can clear the data in the information_schema.PIPELINE_FILES table. It stores data for the duration of a pipeline’s lifetime, resulting in a large number of files being referenced in the view. If the data in this view is not required, you can:

  1. To drop a single file and its file record from the information_schema.PIPELINES_FILES table, use the ALTER PIPELINE ... DROP FILE command. For example:

    ALTER PIPELINE <your_pipeline> DROP FILE 'pipeline_files';
  2. To drop all of the files associated with a pipeline, you can drop and recreate the pipeline. This clears all the file records associated with the pipeline. However, the metadata related to the pipeline will also be deleted. If you do need the metadata, copy it to a separate table before recreating the pipeline. To recreate the pipeline:

    1. Inspect and note the pipeline settings to use when recreating the pipeline.

      SHOW CREATE PIPELINE <your_pipeline> EXTENDED;
    2. Stop the pipeline.

      STOP PIPELINE <your_pipeline>;
    3. Drop the pipeline.

      DROP PIPELINE <your_pipeline>;
    4. Recreate the pipeline using the required configuration options.

    5. Start the pipeline.

      START PIPELINE <your_pipeline>;

Last modified: October 8, 2024

Was this article helpful?