<jupyter_start><jupyter_text>CLX WorkflowThis is an introduction to the CLX Workflow and it's I/O components.  What is a CLX Workflow?A CLX Workflow receives data from a particular source, performs operations on that data within a GPU dataframe, and outputs that data to a particular destination. This guide will teach you how to configure your workflow inputs and outputs around a simple workflow example.  When to use a CLX WorkflowA CLX Workflow provides a simple and modular way of "plugging in" a particular workflow to a read from different inputs and outputs. Use a CLX Workflow when you would like to deploy a workflow as part of a data pipeline.  A simple example of a custom Workflow<jupyter_code>from clx.workflow.workflow import Workflow
class CustomWorkflow(Workflow):
    def workflow(self, dataframe):
        dataframe["enriched"] = "enriched output"
        return dataframe<jupyter_output><empty_output><jupyter_text>The Workflow relies on the Workflow class which handles the I/O and general data processing functionality. To implement a new Workflow, the developer need only implement the `workflow` function which receives an input dataframe, as shown above.   A more advanced example of a Worlflow can be found [here](https://github.com/rapidsai/clx/blob/branch-0.12/clx/workflow/splunk_alert_workflow.py).It is an example of a [Splunk](https://www.splunk.com/) Alert Workflow used to find anamolies in Splunk alert data.  Workflow I/O ComponentsIn order to deploy a workflow to an input an output data feed, we integrate the CLX I/O components. Let's look quickly at what a workflow configuration for the source and destination might look like. You can see below we declare each of the properties within a dictionary. For more information on how to declare configuration within a configurable yaml file [go].<jupyter_code>source = {
   "type": "fs",
   "input_format": "csv",
   "input_path": "/full/path/to/input/data",
   "schema": ["raw"],
   "delimiter": ",",
   "required_cols": ["raw"],
   "dtype": ["str"],
   "header": 0
}
destination = {
   "type": "fs",
   "output_format": "csv",
   "output_path": "/full/path/to/output/data"
}<jupyter_output><empty_output><jupyter_text>The first step to configuring the input and output of a workflow is to determine the source and destination type. Then to set the associated parameters for that specific type.As seen above the `type` property is listed first and can be one of the following. Source Types  * `fs` - Read from a local filesystem* `dask_fs` - Increase the speed of GPU workflow operations by reading from a file using Dask* `kafka` - Read from [Kafka](https://kafka.apache.org/)Destination Types  * `fs` - Writing to local filesystem* `kafka` - Write to [Kafka](https://kafka.apache.org/)  Source and Destination Configurations FilesystemIf the `fs` type is used, the developer must distinguish the data format using the `input_format` attribute. Formats available are: csv, parquet, and orc.  The associated parameters available for the `fs` type and `input_format` are documented within the [cuDF I/O](https://rapidsai.github.io/projects/cudf/en/0.11.0/api.htmlmodule-cudf.io.csv) API. For example for reading data from a csv file, reference [cudf.io.csv.read_csv](https://rapidsai.github.io/projects/cudf/en/0.11.0/api.htmlcudf.io.csv.read_csv) available parameters.Example<jupyter_code>source = {
   "type": "fs",
   "input_format": "parquet",
   "input_path": "/full/path/to/input/data",
   "columns": ["x"]
}<jupyter_output><empty_output><jupyter_text>Dask Filesystem  If the `dask_fs` type is used the developer must distinguish the data format using the `input_format` attribute. Formats available are: csv, parquet, and orc.  The associated parameters available for the `dask_fs` type and `input_format` are listed within the [Dask cuDF](https://rapidsai.github.io/projects/cudf/en/0.11.0/10min.htmlGetting-Data-In/Out) documentation.   Example<jupyter_code>source = {
   "type": "dask_fs",
   "input_format": "csv",
   "input_path": "/full/path/to/input/data/*.csv"
}<jupyter_output><empty_output><jupyter_text>KafkaIf the `kafka` type is used the following parameters must be indicated    Source    * `kafka_brokers` - Kafka brokers* `group_id` - Group ID for consuming kafka messages* `consumer_kafka_topics` - Names of kafka topics to read from* `batch_size` - Indicates number of kafka messages to read before data is processed through the workflow* `time_window` - Maximum time window to wait for `batch_size` to be reached before workflow processing begins.   Destination    * `kafka_brokers` - Kafka brokers* `publisher_kafka_topic` - Names of kafka topic to write data to* `batch_size` - Indicates number of workflow-processed messages to aggregate before data is written to the kafka topic* `output_delimiter` - Delimiter of the data columns  Example<jupyter_code>source = {
    "type": "kafka",
    "kafka_brokers": "kafka:9092",
    "group_id": "cyber",
    "batch_size": 10,
    "consumer_kafka_topics": ["topic1", "topic2"],
    "time_window": 5
}
dest = {
    "type": "kafka",
    "kafka_brokers": "kafka:9092"
    "batch_size": 10,
    "publisher_kafka_topic": "topic3",
    "output_delimiter": ","
}<jupyter_output><empty_output><jupyter_text>Tying it together Once we have established our workflow and source and destination configurations we can now run our workflow. Let's create a workflow using the `CustomWorkflow` we created above.Firstly, we must know the parameters for instantiating a basic workflow  * `name` - The name of the workflow* `source` - The source of input data (optional)* `destination` - The destination for output data (optional)<jupyter_code>from clx.workflow.workflow import Workflow
class CustomWorkflow(Workflow):
    def workflow(self, dataframe):
        dataframe["enriched"] = "enriched output"
        return dataframe
    
source = {
   "type": "fs",
   "input_format": "csv",
   "input_path": "/full/path/to/input/data",
   "schema": ["raw"],
   "delimiter": ",",
   "required_cols": ["raw"],
   "dtype": ["str"],
   "header": 0
}
destination = {
   "type": "fs",
   "output_format": "csv",
   "output_path": "/full/path/to/output/data"
}

my_new_workflow = CustomWorkflow(source=source, destination=destination, name="my_new_workflow")
my_new_workflow.run_workflow()<jupyter_output><empty_output><jupyter_text>Workflow configurations in an external fileSometimes workflow configurations may need to change dependent upon the environment. To avoid declaring workflow configurations within sourcecode you may also declare them in an external yaml file. A workflow will look for and establish I/O connections by searching for configurations in the following order:  1. /etc/clx/[workflow-name]/workflow.yaml1. ~/.config/clx/[workflow-name]/workflow.yaml1. In-line python config  If source and destination are indicated in external files, they are not required to instantiate a new workflow<jupyter_code># Workflow config located at /etc/clx/my_new_workflow/workflow.yaml
my_new_workflow = CustomWorkflow(name="my_new_workflow")<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.0995494226978316, 'fraction_numerical': 0.005350605463249789, 'mean_word_length': 4.836483155299918, 'pattern_counts': {'":': 40, '<': 29, '<?xml version=': 0, '>': 29, 'https://': 7, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '7', 'dir': 'jupyter-structured-clean-dedup', 'id': '3441', 'license': 'Apache-2.0', 'n_tokens_mistral': 2155, 'n_tokens_neox': 2081, 'n_words': 775, 'path': 'docs/source/intro-clx-workflow.ipynb', 'repo_name': 'jakirkham/clx', 'size': '7102'}