Flow Configuration - VladimirSiv/pytransflow GitHub Wiki

pytransflow allows flow level configuration, these parameters will influence how the flow is executed and how the flow handles different fail scenarios.

The following sections showcase available flow configuration.

Instant Fail

Parameter instant_fail is optional, defaults to False. If set to True, the flow will throw pytransflow.exceptions.flow.FlowInstantFailException exception if any failure is noticed in the pipeline during the processing. Essentially, it won't allow failed records.

For example, the following flow is using the instant_fail: False

transformations:
  - add_field:
      name: a
      value: 1

If this flow is executed with input records = [{"b": "1"}, {"a": "2"}]. We get the following resulting dataset

Result: {'default': [{'a': 1}]}
Failed Records: [{'failed_records': [FailedRecord(record={'a': '2'}, ...)], ...}]

Record {"a": 1} failed because the transformation add_field is trying to add a field that already exists. However, that didn't stop the flow from moving to the next record and storing the failed one in failed_records.

On the other hand, if we instant_fail: True

instant_fail: True
transformations:
  - add_field:
      name: a
      value: 1

And run it again, we will get the following traceback

...
pytransflow.exceptions.flow.FlowPipelineInstantFailException: Flow Pipeline
raised the instant fail in the flow caused by the error:
OutputAlreadyExistsException(Output field 'a' already exists)

The above exception was the direct cause of the following exception:
...
pytransflow.exceptions.flow.FlowInstantFailException: Flow raised instant fail exception

Fail Scenarios

pytransflow keeps flow statistics during the execution of a flow. Fail scenarios define cases when we want to intentionally fail a flow based on some metric.

The following sections describe some fail scenarios.

Percentage of Failed Records

fail_scenarios:
  percentage_of_failed_records: 50
transformations:
  - add_field:
      name: a
      value: 1

This example defines a threshold of 50 percent for failed records. If the actual percentage is greater or equal to this value, the flow will fail by raising FlowFailScenarioException exception.

For example, using the above flow and the following records:

records = [{"b": "1"}, {"a": "2"}, {"a": 3}]

Will raise the following exception:

pytransflow.exceptions.flow.FlowFailScenarioException:
Flow Fail Scenario 'percentage_of_failed_records': Threshold defined as 50, got 67

As expected, 2 of 3 records fail which results in surpassing the threshold.

Number of Failed Records

Same as the previous example, but instead of percentage we are defining the rule for number of failed records.

fail_scenarios:
  number_of_failed_records: 2
transformations:
  - add_field:
      name: a
      value: 1

In this example, if the number of failed records is equal or greater than 2, the flow will fail, raising the FlowFailScenarioException exception.

Running this flow using the following records

records = [{"b": "1"}, {"a": "2"}, {"a": 3}]

will raise the following exception

pytransflow.exceptions.flow.FlowFailScenarioException: Flow Fail Scenario
'number_of_failed_records': Threshold defined as 2, got 2

Datasets Present

This flow failure scenario is checking if a dataset is present in the final dataset result. If it's present, it will raise the error. The configuration parameter takes a list of strings, which represent dataset names.

fail_scenarios:
  datasets_present: ["a"]
transformations:
  - add_field:
      name: a
      value: 1
      output_datasets: ["a"]

Running this flow with an empty record will raise the following exception

pytransflow.exceptions.flow.FlowFailScenarioException: Flow Fail Scenario
'datasets_present': Dataset 'a' is present

Datasets Not Present

fail_scenarios:
  datasets_not_present: ["a"]
transformations:
  - add_field:
      name: a
      value: 1

Running this flow with an empty record will raise the following exception

pytransflow.exceptions.flow.FlowFailScenarioException: Flow Fail Scenario
'datasets_not_present': Dataset 'a' is not present

Parallelization

pytransflow supports running flows in multiprocessing mode i.e. in parallel. This is achieved by creating multiple processes where each process is running a single flow pipeline and works on a batch of records. For more information on how it works, please see How it Works.

Enabling parallelization is done via parallel parameter, which defaults to False.

parallel: True
transformations:
  - add_field:
      name: a
      value: 1

Additional options include:

  • batch - Batch size i.e. number of records in a batch. If not specified, size of a batch will be equal to number of input records divided by number of available cores. Defaults to None.
  • cores - Number of cores, if not specified it will take the number of available cores in the system. Defaults to None.

For example:

parallel: True
batch: 20
cores: 4
transformations:
  - add_field:
      name: a
      value: 1

Flow Variables

Defining variables on a flow level is supported. These variables can be used in any transformation, particularly in condition expressions. The idea is to define a value at the beginning of the flow and just reference it where you need it, using !: pattern.

variables:
  d: "A"
transformations:
  - add_field:
      name: b
      value: 1
      condition: "@a == !:d"

Running this flow with the following input records:

records = [{"a": "A"}]

Outputs

Result: {'default': [{'a': 'A', 'b': 1}]}

However, changing it to {"a": "<something-else>"}, gives

Result: {'default': [{'a': 'a'}]}

In other words, condition for transforming the record is not met.

Note: Flow variables can be created, deleted, or modified from a transformation, if you ever wish to do so. This allows you to dynamically set flow variables. To do so, you'll have to create a custom transformation, since this is not done in built-in transformations. Please see Custom Transformations for more details.