Streaming in Mule

Objectives

• Streaming Overview
• Advantages of Streaming
• Limitations
• Use cases
• Streaming with Sample workflow
• Testing Metrics and execution report

Mule 4 Testing Metrics with 1 million ,10 million records.

Mule 3 Testing Metrics with 1M,10M and 20M records.

Streaming Overview

Streaming is the process where we just refer to the data as its bytes arrive instead of scanning and loading the entire document to index it. Streaming speeds up the processing of large documents without overloading memory. Mule supports end-to-end streaming throughout a flow.

Enablement of Streaming 

We can enable streaming through 2 modes:

1.  Through internal configuration properties of streaming inside connectors configurations. Example: opting for streaming strategy properties
2.  Through data weave configuration properties, we can achieve streaming output. Example: deferred=true or streaming=true

Types of Streaming 

We have the option to set streaming strategy configuration inside the connector, which has three types of streaming:

1. Repeatable file store stream (default): Mule 4 introduces repeatable streams as its default framework handling streams. It enables us to read stream data more than once and have concurrent access to the stream.
2.  Non-repeatable stream:
   1. The stream can only be read once.
   2. Non-repeatable streaming improves performance.
   3. Need a very tight optimization for performance and resource consumption.
3. Repeatable in-memory stream: This strategy creates a temporary file to the disk to store the contents without allocated buffer size. It is useful to use for small sizes of data.
   1. Data Format that supports streaming: CSV, JSON, XML
   2. Connectors support streaming: File, FTP, SFTP, Database, HTTP, etc.

Advantages of Streaming 

• The streaming technique offers a huge advantage over fully loading the data into memory since it prevents the Java heap from being exhausted by big input files. With less allocated heap size, we can execute large data sets.
• Batch execution and other execution scopes for processing large datasets can be removed, which can reduce code complexity.
• Based on our previous PSR result, we have observed that the throughput has increased 30x times (3000 Rec/sec Vs 90000 Rec/sec)
• The execution time difference from the previous test result to streaming goes down from 1.5Hr to 3.35 Min for 20M records.
• Increase of max streaming memory to process huge data load in the wrapper config file.

Limitations

• In non-repeatable streaming, streamed data can be referenced only once. No reference to the streamed payload is found in a nested lambda. Access of output Streamed data fails if a script attempts to reference the same payload more than once with a pipe closed error.
• Exception handling to capture faulty data from the streamed payload is not achievable.
• If the stream exceeds this limit, the application fails. This strategy defaults to a buffer size of 512 KB. For larger streams, the buffer is expanded by a default increment size of 512 KB until it reaches the configured maximum buffer size.
• Grouping logic is not supported.
• The lookup function will not work with deferred = true.

Use Cases

• When data validation is not required
• When business logic is less
• When the data load is huge to process

Streaming With Sample Workflow 

Mule 4 Observation:

Mule Runtime: 4.4.0 

We have created a prototype application to understand the e2e streaming with a very basic sample flow: 

• Full Streaming: Create a sample basic flow and try to read an input CSV file through streaming and convert it into JSON format with streaming and write the content into JSON
  file via the local file system.
• Partial Streaming: Create a sample basic flow and try to read an input CSV file with the non-repeatable streaming method and convert the data into JSON format with streaming and write it into a local file system in chunks of 1000 size. We can observe here that
  before writing the payload into the file, we are using for each to divide the data into chunks which will load the data into memory, thus describing a partial streaming strategy.

Flow: 

  Testing Metrics: 

1. Partial Streaming Metrics:

• 10M Records
• Streaming: Partial
• Heap size: 4GB
• Process timing: 18 minutes

2. Full Streaming Metrics:

• 10M Records
• Heap Size: 1GB
• Streaming: Yes
• Process timing: 3 minutes
• Testing Metrics:

Code:

Execution Table Metrics Report: 

Mule 3 Observations:

Mule Runtime: 4.3.0

Create a prototype application to understand the e2e streaming with basic sample flow. Try to read an input CSV file through on new or updated file connector in repeatable streaming mode as we want to refer the file in concurrent method with scatter-gather.
Use a set payload connector for setting a mime-type with streaming properties to achieve octet-stream or byte streaming with proper specification of mime-type for an ingested input file and add it as an additional property. After that process, the input file
concurrently with scatter-gather to get two formats of data, one is for XML format, and the other is for JSON format, and then write it into a file via the local file system.

• Property configuration of On New or Update connector to read input file in repeatable stream mode.

• Property configuration of set Payload connector.

• Process XML dwl file.

• Process JSON dwl file.

Output:

Execution time of JSON: 0:54:00 for 1M Item records

Execution time of XML: 0:01:09 for 1M Item records

Streaming With Grouping Logic

Observation: 

We have tested a few entities in the Adapter which have grouping logic like Item, Actual vehicle load outbound, Production method, etc., where we have processed the data with grouping logic like grouping the data based on a few columns ex: material number or Item number, etc.
which will be applicable to the whole ingested payload. In this case, the Input payload will get loaded into memory at the time of DWL execution due to group by function, and then further transformation will get processed, which can result in back pressure in case of a huge payload that causes the application to restart abruptly or may cause more resource utilization.
Below is the Testing report:

Streaming Without Grouping Logic

Observation:

We have also tested entities that do not have grouping logic. So, we have tested one of the entities for millions of records on the Adapter and captured a few test results below:  

Observation on Existing Issue

Streaming role for faulty records or error records: 

1. In case of any error, the message event is not getting redirected to the error handler.

Mulesoft Product team Response:

When we use deferred=true, it will pass the stream to the next component without directly throwing the error on the Transform Message component. This is the downside of using deferred. This is expected behavior.

2. In case of any error, the first faulty record, and subsequent records get dropped.

Mulesoft Product team Response:

All the successful records are getting processed, and error records are getting skipped. Expected behavior from Mulesoft.

3. Handle error records such that they can be pushed to an API.

Mulesoft Product team Response:

If we tried to pass failed records information to another flow using the lookup function, which cannot be used with deferred mode true. Expected behavior from Mulesoft

4. Handle exceptions for error records so that exception information can be logged and stored as per business use case.

Mulesoft Product team Response:

Able to successfully log the erroneous records to the console with a custom error message from the data weave code itself.