Koheesio

https://github.com/nike-inc/koheesio
Admin
Koheesio is a robust Python framework designed to build efficient data pipelines by encouraging...
Tokens:69,301
Snippets:556
Trust Score:9.4
License:Apache-2.0
Update:1 week ago
Show doc for...
Tokens:
Raw
### Install and start Podman

Source: https://github.com/nike-inc/koheesio/blob/main/docs/development/local-github-actions.md

Install Podman via Homebrew, initialize and start the machine. Fully open source alternative.

```bash
brew install podman
podman machine init
podman machine start
```

--------------------------------

### Install act using Make

Source: https://github.com/nike-inc/koheesio/blob/main/docs/development/local-github-actions.md

Recommended installation method. Automatically detects platform and installs act.

```bash
make act-install
```

--------------------------------

### Install and start Colima

Source: https://github.com/nike-inc/koheesio/blob/main/docs/development/local-github-actions.md

Install Colima and Docker CLI via Homebrew, then start Colima with 4 CPUs and 8GB memory. Recommended for macOS/Linux.

```bash
# Install
brew install colima

# Install Docker CLI (Colima requires it)
brew install docker

# Start Colima
colima start --cpu 4 --memory 8
```

--------------------------------

### Start container runtime for Colima, Docker Desktop, Podman, OrbStack

Source: https://github.com/nike-inc/koheesio/blob/main/docs/development/local-github-actions.md

Solutions for the error 'Container runtime is not running'. Start the appropriate container runtime for your setup.

```bash
# Colima
colima start

# Docker Desktop
# Start the Docker Desktop application

# Podman
podman machine start

# OrbStack
# Start the OrbStack application
```

--------------------------------

### Install hatch with make init

Source: https://github.com/nike-inc/koheesio/blob/main/docs/community/contribute.md

Use `make init` to install hatch, the required package manager. Do not use pip or conda for dependency installation.

```console
make init
```

--------------------------------

### Manual installation on Linux

Source: https://github.com/nike-inc/koheesio/blob/main/docs/development/local-github-actions.md

Install act via curl script on Linux.

```bash
curl https://raw.githubusercontent.com/nektos/act/master/install.sh | sudo bash
```

--------------------------------

### Install act with make act-install

Source: https://github.com/nike-inc/koheesio/blob/main/docs/development/local-github-actions.md

Solution for the error 'act is not installed'. Install act using the provided make target or manually.

```bash
make act-install
# Or install manually (see Installation section above)
```

--------------------------------

### Set up dev environment with make dev

Source: https://github.com/nike-inc/koheesio/blob/main/CONTRIBUTING.md

Installs the required packages for the development environment. Run after installing hatch.

```bash
make dev
```

--------------------------------

### Manual installation on macOS

Source: https://github.com/nike-inc/koheesio/blob/main/docs/development/local-github-actions.md

Install act via Homebrew on macOS.

```bash
brew install act
```

--------------------------------

### Example: Step Documentation with NumPy-style Docstring

Source: https://github.com/nike-inc/koheesio/blob/main/docs/community/approach-documentation.md

Shows a complete Koheesio Reader class with a NumPy-style docstring including parameters, notes, examples, and see also sections. The example demonstrates how to document a step that reads data from an external source into a Spark DataFrame, supporting both batch and streaming modes.

```python
class MyReader(Reader):
    """Reads data from an external source.

    This reader connects to an external data source and loads data into a Spark DataFrame.
    It supports both batch and streaming modes.

    Parameters
    ----------
    source_path : str
        Path to the data source
    format : str
        Data format (e.g., 'json', 'csv', 'parquet')
    options : Optional[Dict[str, Any]]
        Additional options for the reader

    Notes
    -----
    * Make sure the necessary dependencies are available in your Spark session
    * For streaming mode, set `streaming=True`

    Examples
    --------
    ```python
    from koheesio.spark.readers import MyReader

    reader = MyReader(
        source_path="s3://bucket/data",
        format="parquet"
    )
    df = reader.execute().df
    ```

    See Also
    --------
    * [Reader](../reference/reader.md): Base reader class
    * [Writer](../reference/writer.md): For writing data
    """

    source_path: str = Field(default=..., description="Path to the data source")
    format: str = Field(default="parquet", description="Data format to read")
```

--------------------------------

### Install hatch with make hatch-install

Source: https://github.com/nike-inc/koheesio/blob/main/docs/community/contribute.md

Alternative command to install hatch using `make hatch-install`. On macOS, this installs hatch via brew.

```console
make hatch-install
```

--------------------------------

### List available make commands

Source: https://github.com/nike-inc/koheesio/blob/main/docs/community/contribute.md

Run `make help` to see the full list of available make commands for managing setup and maintenance tasks.

```bash
make help
```

--------------------------------

### Install OrbStack

Source: https://github.com/nike-inc/koheesio/blob/main/docs/development/local-github-actions.md

Install OrbStack via Homebrew. macOS alternative with fast startup and modern UI.

```bash
brew install orbstack
```

--------------------------------

### Install act via winget

Source: https://github.com/nike-inc/koheesio/blob/main/docs/development/local-github-actions.md

Installs act using the Windows package manager winget.

```bash
winget install nektos.act
```

--------------------------------

### Install act via Scoop

Source: https://github.com/nike-inc/koheesio/blob/main/docs/development/local-github-actions.md

Installs act using the Scoop package manager.

```bash
scoop install act
```

--------------------------------

### Build and run an ETL task with Spark

Source: https://github.com/nike-inc/koheesio/blob/main/docs/tutorials/getting-started.md

Advanced example showing how to set up a SparkSession, configure a Context with readers, transformations, and writers, define an EtlTask subclass, and run it. The task adds a literal column using a context value.

```python
from pyspark.sql.functions import lit
from pyspark.sql import DataFrame, SparkSession

# Step 1: import Koheesio dependencies
from koheesio.context import Context
from koheesio.spark.readers.dummy import DummyReader
from koheesio.spark.transformations.renames.camel_to_snake import CamelToSnakeTransformation
from koheesio.spark.writers.dummy import DummyWriter
from koheesio.spark.etl_task import EtlTask

# Step 2: Set up a SparkSession
spark = SparkSession.builder.getOrCreate()

# Step 3: Configure your Context
context = Context({
    "source": DummyReader(),
    "transformations": [CamelToSnakeTransformation()],
    "target": DummyWriter(),
    "my_favorite_movie": "inception",
})

# Step 4: Create a Task
class MyFavoriteMovieTask(EtlTask):
    my_favorite_movie: str

    def transform(self, df: DataFrame = None) -> DataFrame:
        df = df.withColumn("MyFavoriteMovie", lit(self.my_favorite_movie))
        return super().transform(df)

# Step 5: Run your Task
task = MyFavoriteMovieTask(**context)
task.run()
```

--------------------------------

### Install hatch with make init hatch-install

Source: https://github.com/nike-inc/koheesio/blob/main/CONTRIBUTING.md

Installs the `hatch` package manager on macOS. This is required before setting up the dev environment. Do not use pip or conda to install dependencies.

```bash
make init hatch-install
```

--------------------------------

### Install Koheesio with pip

Source: https://github.com/nike-inc/koheesio/blob/main/README.md

Use this command to install the Koheesio package via pip.

```bash
pip install koheesio
```

--------------------------------

### Getting Step Description

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/concepts/step.md

Shows how to get the description of a step using the get_description method.

```APIDOC
## Getting Step Description

### Description
You can get the description of a step using the `get_description` method.

### Method
`get_description()`

### Code Example
```python
step = MyStep(input1="value1", input2=2)
description = step.get_description()
```
```

--------------------------------

### Install act via Chocolatey

Source: https://github.com/nike-inc/koheesio/blob/main/docs/development/local-github-actions.md

Installs act using the Chocolatey package manager.

```bash
choco install act-cli
```

--------------------------------

### Field Documentation Example with Pydantic Fields

Source: https://github.com/nike-inc/koheesio/blob/main/docs/community/approach-documentation.md

Shows how to document Pydantic fields in a Koheesio Step class using the description parameter. The example includes an input column and a threshold field with their descriptions.

```python
class MyStep(Step):
    input_column: str = Field(
        default=...,
        description="Name of the input column to process"
    )
    threshold: int = Field(
        default=100,
        description="Threshold value for filtering"
    )
```

--------------------------------

### Use HelloWorldStep with optional spark session

Source: https://github.com/nike-inc/koheesio/blob/main/docs/tutorials/hello-world.md

Instantiates HelloWorldStep with a message and optionally passes a spark session. After execution, the output DataFrame is accessed via output.df and shown. If no spark session is passed, Koheesio attempts to get an active session.

```python
hello_world_step = HelloWorldStep(message="Hello, World!", spark=spark)  # optionally pass the spark session
hello_world_step.execute()

hello_world_step.output.df.show()
```

--------------------------------

### Get Step description with get_description

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/concepts/step.md

Call get_description on a step instance to retrieve its description.

```python
step = MyStep(input1="value1", input2=2)
description = step.get_description()
```

--------------------------------

### Using the .read() method

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/spark/readers.md

Shows how to instantiate a Reader subclass and call .read() to get a DataFrame. The .read() method is a shorthand for .execute().output.df.

```python
# Instantiate the Reader
my_reader = MyReader()

# Use the .read() method to get the data as a DataFrame
df = my_reader.read()

# Now df is a DataFrame with the data read by MyReader
```

--------------------------------

### Access multiple Step outputs

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/concepts/step.md

After execution, access each output via step.output.output_name. The example prints both outputs.

```python
step = MyStep(input1="value1", input2=2, input3=3)
step.execute()
print(step.output.output1)  # Outputs: "result"
print(step.output.output2)  # Outputs: 5
```

--------------------------------

### Install Hatch conditionally for act vs GitHub Actions

Source: https://github.com/nike-inc/koheesio/blob/main/docs/development/local-github-actions.md

YAML steps that install Hatch via pip when running in act (ACT env var set) and via the pypa/hatch action in GitHub Actions.

```yaml
- name: Install Hatch (for local act)
  if: ${{ env.ACT }}
  run: pip install hatch

- name: Install Hatch (for GitHub Actions)
  if: ${{ !env.ACT }}
  uses: pypa/hatch@install
```

--------------------------------

### Using a Context class to configure an EtlTask

Source: https://github.com/nike-inc/koheesio/blob/main/docs/tutorials/onboarding.md

Shows how to define a Context object with reader, writer, and transformation options, then pass them to an EtlTask. The Context behaves like a dictionary and can be loaded from JSON or YAML. The example filters data based on an age threshold.

```python
from koheesio import Context
from koheesio.spark.etl_task import EtlTask
from koheesio.spark.readers.file_loader import CsvReader
from koheesio.spark.writers.delta import DeltaTableWriter
from koheesio.spark.transformations.transform import Transform

context = Context({  # this could be stored in a JSON or YAML
    "age_threshold": 18,
    "reader_options": {
        "path": "input.csv",
        "header": True,
        "inferSchema": True
    },
    "writer_options": {
        "table": "delta_table",
        "partition_by": ["country"]
    }
})

task = EtlTask(
    source = CsvReader(**context.reader_options),
    target = DeltaTableWriter(**context.writer_options),
    transformations = [
        Transform(func=lambda df: df.filter(df["age"] > context.age_threshold))
    ]
)

task.execute()
```

--------------------------------

### Access Step output after execution

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/concepts/step.md

After calling execute, access the output via step.output.output_name. The example prints the output value.

```python
step = MyStep(input1="value1", input2=2)
step.execute()
print(step.output.output1)  # Outputs: "result"
```

--------------------------------

### Define a Transformation subclass with execute method

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/spark/transformations.md

Create a subclass of Transformation and implement execute to read from self.input.df, apply a transformation, and store the result in self.output.df. The example uses an undefined apply_transformation function.

```python
class MyTransformation(Transformation):
    def execute(self):
        # get data from self.input.df
        data = self.input.df
        # apply transformation
        transformed_data = apply_transformation(data)
        # store result in self.output.df
        self.output.df = transformed_data
```

--------------------------------

### Access data via the df property

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/spark/writers.md

Shows how to use the df property to get the data as a DataFrame. This is a shorthand for self.input.df and ensures data is ready even if execute hasn't been called.

```python
# Instantiate the Writer
my_writer = MyWriter()

# Use the df property to get the data as a DataFrame
df = my_writer.df

# Now df is a DataFrame with the data that will be written by MyWriter
```

--------------------------------

### Serialize a Step to YAML with to_yaml

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/concepts/step.md

Call to_yaml on a step instance to get a YAML string representation.

```python
step = MyStep(input1="value1", input2=2)
yaml_str = step.to_yaml()
```

--------------------------------

### DataframeLookup transformation example

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/spark/transformations.md

Joins two DataFrames using DataframeLookup with a join mapping, target column, and left join type. Requires a SparkSession and koheesio.spark.transformations.lookup imports. Executes the transformation and retrieves the output DataFrame.

```python
from pyspark.sql import SparkSession
from koheesio.spark.transformations.lookup import DataframeLookup, JoinMapping, TargetColumn, JoinType

spark = SparkSession.builder.getOrCreate()
left_df = spark.createDataFrame([(1, "A"), (2, "B")], ["id", "value"])
right_df = spark.createDataFrame([(1, "A"), (3, "C")], ["id", "value"])

lookup = DataframeLookup(
    df=left_df,
    other=right_df,
    on=JoinMapping(source_column="id", other_column="id"),
    targets=TargetColumn(target_column="value", target_column_alias="right_value"),
    how=JoinType.LEFT,
)

output_df = lookup.execute().df
```

--------------------------------

### Define a custom SparkStep

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/concepts/step.md

Example of a SparkStep subclass with an input field and an Output class containing a DataFrame. The execute method reads a text file using the SparkSession and assigns the result to the output.

```python
class MySparkStep(SparkStep):
    input1: str = Field(...)
    
    class Output(StepOutput):
        output1: DataFrame = Field(...)
    
    def execute(self):
        # Your logic here
        df = self.spark.read.text(self.input1)
        self.output.output1 = df
```

--------------------------------

### Custom metaclass overriding _log_end_message and _validate_output

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/concepts/step.md

Python examples showing how to create custom metaclasses by subclassing StepMetaClass and overriding _log_end_message or _validate_output. The first overrides _log_end_message to print a custom message, the second overrides _validate_output to always set a dummy value in the step output. These are used as metaclasses for Step subclasses.

```python
    class MyMetaClass(StepMetaClass):
        @classmethod
        def _log_end_message(cls, step: Step, skip_logging: bool = False, *args, **kwargs):
            print("It's me from custom meta class")
            super()._log_end_message(step, skip_logging, *args, **kwargs)

    class MyMetaClass2(StepMetaClass):
        @classmethod
        def _validate_output(cls, step: Step, skip_validating: bool = False, *args, **kwargs):
            # i want always have a dummy value in the output
            step.output.dummy_value = "dummy"

    class YourClassWithCustomMeta(Step, metaclass=MyMetaClass):
        def execute(self):
            self.log.info(f"This is from the execute method of {self.__class__.__name__}")

    class YourClassWithCustomMeta2(Step, metaclass=MyMetaClass2):
        def execute(self):
            self.log.info(f"This is from the execute method of {self.__class__.__name__}")
```

--------------------------------

### Unit test an EtlTask with a filter transformation

Source: https://github.com/nike-inc/koheesio/blob/main/docs/tutorials/testing-koheesio-steps.md

This example shows how to unit test a Koheesio EtlTask. It creates a Spark DataFrame, defines a filter function, and asserts that the task correctly filters out rows where age is less than or equal to 18.

```python
from pyspark.sql import SparkSession
from pyspark.sql.functions import col

from koheesio.spark import DataFrame
from koheesio.spark.etl_task import EtlTask
from koheesio.spark.readers.dummy import DummyReader
from koheesio.spark.writers.dummy import DummyWriter
from koheesio.spark.transformations.transform import Transform


def filter_age(df: DataFrame) -> DataFrame:
    return df.filter(col("Age") > 18)


def test_etl_task():
    # Initialize SparkSession
    spark = SparkSession.builder.getOrCreate()

    # Create a DataFrame for testing
    data = [("John", 19), ("Anna", 20), ("Tom", 18)]
    df = spark.createDataFrame(data, ["Name", "Age"])

    # Define the task
    task = EtlTask(
        source=DummyReader(df=df),
        target=DummyWriter(),
        transformations=[
            Transform(filter_age)
        ]
    )

    # Execute the task
    task.execute()

    # Assert the result
    result_df = task.output.df
    assert result_df.count() == 2
    assert result_df.filter("Name == 'Tom'").count() == 0
```

--------------------------------

### Define and run a basic Step

Source: https://github.com/nike-inc/koheesio/blob/main/docs/tutorials/getting-started.md

Shows how to create a custom Step subclass and execute it. The execute method is where you place your step logic.

```python
from koheesio import Step

# Define a step
class MyStep(Step):
    def execute(self):
        # Your step logic here

# Create an instance of the step
step = MyStep()

# Run the step
step.execute()
```

--------------------------------

### sample.yaml Configuration

Source: https://github.com/nike-inc/koheesio/blob/main/docs/tutorials/hello-world.md

Configuration file providing context for the task, including source data settings and the favorite movie title.

```yaml
raw_layer:
  catalog: development
  schema: my_favorite_team
  table: some_random_table
movies:
  favorite: Office Space
hash_settings:
  source_columns:
  - id
  - foo
  target_column: hash_uuid5
source:
  range: 4
```

--------------------------------

### Use the .write() method to write data

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/spark/writers.md

Shows how to instantiate a Writer and call .write() to write the data. This is a shorthand for calling .execute().output.df.

```python
# Instantiate the Writer
my_writer = MyWriter()

# Use the .write() method to write the data
my_writer.write()

# The data from MyWriter's DataFrame is now written to the destination
```

--------------------------------

### Format codebase with make fmt

Source: https://github.com/nike-inc/koheesio/blob/main/docs/community/contribute.md

Run `make fmt` to format the codebase using black, isort, and ruff. Ensure no formatting issues remain before submitting a pull request.

```bash
make fmt
```

--------------------------------

### Get Independent Logger

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/concepts/logger.md

Get an independent logger without inheriting from the LoggingFactory logger by setting inherit to False.

```python
your_logger = factory.get_logger(name="your_logger_name", inherit=False)
```

--------------------------------

### Instantiate and Execute MyFavoriteMovieTask

Source: https://github.com/nike-inc/koheesio/blob/main/docs/tutorials/hello-world.md

Loads the context from sample.yaml, creates a SparkSession, and instantiates MyFavoriteMovieTask with a DummyReader source, DummyWriter target, and the favorite movie from context, then executes the task.

```python
from pyspark.sql import SparkSession
from koheesio.context import Context
from koheesio.spark.readers.dummy import DummyReader
from koheesio.spark.writers.dummy import DummyWriter

context = Context.from_yaml("sample.yaml")

SparkSession.builder.getOrCreate()

my_fav_mov_task = MyFavoriteMovieTask(
    source=DummyReader(**context.raw_layer),
    target=DummyWriter(truncate=False),
    my_favorite_movie=context.movies.favorite,
)
my_fav_mov_task.execute()
```

--------------------------------

### Run full test workflow

Source: https://github.com/nike-inc/koheesio/blob/main/docs/development/local-github-actions.md

Runs `make ci-test` to execute all test matrix combinations. Warning: this may take 30+ minutes.

```bash
make ci-test
```

--------------------------------

### Run a Step with multiple inputs

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/concepts/step.md

Instantiate the step with all input values and call execute.

```python
step = MyStep(input1="value1", input2=2, input3=3)
step.execute()
```

--------------------------------

### Running a Step with Multiple Inputs

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/concepts/step.md

Shows how to run a step with multiple inputs.

```APIDOC
## Running a Step with Multiple Inputs

### Description
To run a step with multiple inputs, you can do the following.

### Method
`execute()`

### Code Example
```python
step = MyStep(input1="value1", input2=2, input3=3)
step.execute()
```
```

--------------------------------

### Basic Context Operations

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/concepts/context.md

Shows how to create a Context, access a value, add a key, and merge two Contexts.

```python
# Create a Context
context = Context({"bronze_table": "catalog.schema.table_name"})

# Access a value
value = context.bronze_table

# Add a key
context.add("silver_table", "catalog.schema.table_name")

# Merge two Contexts
context.merge(Context({"silver_table": "catalog.schema.table_name"}))
```

--------------------------------

### Run a quick combination to verify with make ci-test-matrix

Source: https://github.com/nike-inc/koheesio/blob/main/docs/development/local-github-actions.md

Run a quick test matrix combination with Python 3.12 and PySpark 352r to verify the workflow.

```bash
# Run a quick combination to verify
make ci-test-matrix PYTHON=3.12 PYSPARK=352r
```

--------------------------------

### Access the SparkSession via the spark property

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/spark/writers.md

Shows how to use the spark property to get the currently active SparkSession, which can be used for distributed data processing.

```python
# Instantiate the Writer
my_writer = MyWriter()

# Use the spark property to get the SparkSession
spark = my_writer.spark

# Now spark is the SparkSession associated with MyWriter
```

--------------------------------

### Running a Step

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/concepts/step.md

Shows how to run a step by calling the execute method or its alias run.

```APIDOC
## Running a Step

### Description
To run a step, call the `execute` method. The `run` method is an alias to `execute`.

### Method
`execute()` or `run()`

### Code Example
```python
step = MyStep(input1="value1", input2=2)
step.execute()
```
```

--------------------------------

### Getting a key-value pair with get_item method

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/concepts/context.md

Extract a specific key-value pair from a Context using the get_item method. Returns: {"silver_table": "catalog.schema.table_name"}.

```python
context.get_item("silver_table")  # Returns: {"silver_table": "catalog.schema.table_name"}
```

--------------------------------

### Defining a Step with Multiple Inputs and Outputs

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/concepts/step.md

Shows how to define a step with multiple inputs and outputs.

```APIDOC
## Defining a Step with Multiple Inputs and Outputs

### Description
Here's an example of how to define a new step with multiple inputs and outputs.

### Method
Class definition

### Code Example
```python
class MyStep(Step):
    input1: str = Field(...)
    input2: int = Field(...)
    input3: int = Field(...)
    
    class Output(StepOutput):
        output1: str = Field(...)
        output2: int = Field(...)
    
    def execute(self):
        # Your logic here
        self.output.output1 = "result"
        self.output.output2 = self.input2 + self.input3
```
```

--------------------------------

### Verify act and container runtime

Source: https://github.com/nike-inc/koheesio/blob/main/docs/development/local-github-actions.md

Run make act-check to verify both act and your container runtime are properly configured.

```bash
make act-check
```

--------------------------------

### Run a Step using execute

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/concepts/step.md

Instantiate the step with input values and call execute. The run method is an alias for execute.

```python
step = MyStep(input1="value1", input2=2)
step.execute()
```

--------------------------------

### Defining a Step

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/concepts/step.md

Shows how to subclass Step and implement the execute method with inputs and outputs.

```APIDOC
## Defining a Step

### Description
To define a new step, subclass the `Step` class and implement the `execute` method. Inputs are accessed via `self.input_name` and outputs via `self.output.output_name`.

### Method
Class definition

### Code Example
```python
class MyStep(Step):
    input1: str = Field(...)
    input2: int = Field(...)
    
    class Output(StepOutput):
        output1: str = Field(...)
    
    def execute(self):
        # Your logic here
        self.output.output1 = "result"
```
```

--------------------------------

### List available workflows

Source: https://github.com/nike-inc/koheesio/blob/main/docs/development/local-github-actions.md

Runs `make ci-list` to display all workflows and jobs defined in `.github/workflows/`.

```bash
make ci-list
```

--------------------------------

### HashUUID5 transformation example

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/spark/transformations.md

Generates a UUID5 hash for each row based on specified source columns. Requires a SparkSession and koheesio.spark.transformations.uuid5 import. Executes the transformation and stores the hash in a new column named 'hash'.

```python
from pyspark.sql import SparkSession
from koheesio.spark.transformations.uuid5 import HashUUID5

spark = SparkSession.builder.getOrCreate()
df = spark.createDataFrame([(1, "A"), (2, "B")], ["id", "value"])

hash_transform = HashUUID5(
    df=df,
    source_columns=["id", "value"],
    target_column="hash"
)

output_df = hash_transform.execute().df
```

--------------------------------

### Run tests with coverage and all test suites

Source: https://github.com/nike-inc/koheesio/blob/main/docs/community/contribute.md

Run `make cov` to run tests and check coverage, `make all-tests` for all tests, `make spark-tests` for Spark tests, and `make non-spark-tests` for non-Spark tests. All tests must pass with adequate coverage before submitting a pull request.

```bash
make cov  # to run the tests and check the coverage
```

```bash
make all-tests  # to run all the tests
```

```bash
make spark-tests  # to run the spark tests
```

```bash
make non-spark-tests  # to run the non-spark tests
```

--------------------------------

### Using the df property

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/spark/readers.md

Shows how to access the df property, which is a shorthand for self.output.df. If self.output.df is None, the execute method is run first.

```python
# Instantiate the Reader
my_reader = MyReader()

# Use the df property to get the data as a DataFrame
df = my_reader.df

# Now df is a DataFrame with the data read by MyReader
```

--------------------------------

### List all available workflows with act -l

Source: https://github.com/nike-inc/koheesio/blob/main/docs/development/local-github-actions.md

Lists all workflows that act can run locally.

```bash
act -l
```

--------------------------------

### Define a ColumnsTransformation subclass with execute method

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/spark/transformations.md

Create a subclass of ColumnsTransformation and implement execute to add 1 to each column in self.get_columns(). The example imports pyspark.sql.functions and koheesio.spark.transformations.ColumnsTransformation. The ColumnsTransformation class has a ColumnConfig class with fields run_for_all_data_type, limit_data_type, and data_type_strict_mode.

```python
from pyspark.sql import functions as f
from koheesio.spark.transformations import ColumnsTransformation

class AddOne(ColumnsTransformation):
    def execute(self):
        for column in self.get_columns():
            self.output.df = self.df.withColumn(column, f.col(column) + 1)
```

--------------------------------

### Create loggers with LoggingFactory and log messages

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/concepts/logger.md

Shows how to instantiate LoggingFactory, create an independent logger, a logger with a custom level, and child loggers, then log messages at various levels. The comments indicate which messages will be displayed based on the logger's level.

```python
import logging

# Step 2: Instantiate the LoggingFactory class
factory = LoggingFactory(env="local")

# Step 3: Create an independent logger with a custom log level
your_logger = factory.get_logger("your_logger", inherit_from_koheesio=False)
your_logger.setLevel(logging.DEBUG)

# Step 4: Create a logger using the create_logger method from LoggingFactory with a different log level
factory_logger = LoggingFactory(level="WARNING").get_logger(name=factory.LOGGER_NAME)

# Step 5: Create a child logger with a debug level
child_logger = factory.get_logger(name="child")
child_logger.setLevel(logging.DEBUG)

child2_logger = factory.get_logger(name="child2")
child2_logger.setLevel(logging.INFO)

# Step 6: Log messages at different levels for both loggers
your_logger.debug("Debug message")  # This message will be displayed
your_logger.info("Info message")  # This message will be displayed
your_logger.warning("Warning message")  # This message will be displayed
your_logger.error("Error message")  # This message will be displayed
your_logger.critical("Critical message")  # This message will be displayed

factory_logger.debug("Debug message")  # This message will not be displayed
factory_logger.info("Info message")  # This message will not be displayed
factory_logger.warning("Warning message")  # This message will be displayed
factory_logger.error("Error message")  # This message will be displayed
factory_logger.critical("Critical message")  # This message will be displayed

child_logger.debug("Debug message")  # This message will be displayed
child_logger.info("Info message")  # This message will be displayed
child_logger.warning("Warning message")  # This message will be displayed
child_logger.error("Error message")  # This message will be displayed
child_logger.critical("Critical message")  # This message will be displayed

child2_logger.debug("Debug message")  # This message will be displayed
child2_logger.info("Info message")  # This message will be displayed
child2_logger.warning("Warning message")  # This message will be displayed
child2_logger.error("Error message")  # This message will be displayed
child2_logger.critical("Critical message")  # This message will be displayed
```

```sh
[a7d79f7a-f16f-4d2a-8430-1134830f61be] [2023-06-26 20:29:39,267] [DEBUG] [your_logger] {__init__.py:<module>:118} - Debug message
[a7d79f7a-f16f-4d2a-8430-1134830f61be] [2023-06-26 20:29:39,267] [INFO] [your_logger] {__init__.py:<module>:119} - Info message
[a7d79f7a-f16f-4d2a-8430-1134830f61be] [2023-06-26 20:29:39,267] [WARNING] [your_logger] {__init__.py:<module>:120} - Warning message
[a7d79f7a-f16f-4d2a-8430-1134830f61be] [2023-06-26 20:29:39,267] [ERROR] [your_logger] {__init__.py:<module>:121} - Error message
[a7d79f7a-f16f-4d2a-8430-1134830f61be] [2023-06-26 20:29:39,267] [CRITICAL] [your_logger] {__init__.py:<module>:122} - Critical message
[a7d79f7a-f16f-4d2a-8430-1134830f61be] [2023-06-26 20:29:39,267] [WARNING] [koheesio] {__init__.py:<module>:126} - Warning message
[a7d79f7a-f16f-4d2a-8430-1134830f61be] [2023-06-26 20:29:39,267] [ERROR] [koheesio] {__init__.py:<module>:127} - Error message
[a7d79f7a-f16f-4d2a-8430-1134830f61be] [2023-06-26 20:29:39,267] [CRITICAL] [koheesio] {__init__.py:<module>:128} - Critical message
[a7d79f7a-f16f-4d2a-8430-1134830f61be] [2023-06-26 20:29:39,267] [DEBUG] [koheesio.child] {__init__.py:<module>:130} - Debug message
[a7d79f7a-f16f-4d2a-8430-1134830f61be] [2023-06-26 20:29:39,267] [INFO] [koheesio.child] {__init__.py:<module>:131} - Info message
[a7d79f7a-f16f-4d2a-8430-1134830f61be] [2023-06-26 20:29:39,267] [WARNING] [koheesio.child] {__init__.py:<module>:132} - Warning message
[a7d79f7a-f16f-4d2a-8430-1134830f61be] [2023-06-26 20:29:39,267] [ERROR] [koheesio.child] {__init__.py:<module>:133} - Error message
[a7d79f7a-f16f-4d2a-8430-1134830f61be] [2023-06-26 20:29:39,267] [CRITICAL] [koheesio.child] {__init__.py:<module>:134} - Critical message
[a7d79f7a-f16f-4d2a-8430-1134830f61be] [2023-06-26 20:29:39,267] [INFO] [koheesio.child2] {__init__.py:<module>:137} - Info message
[a7d79f7a-f16f-4d2a-8430-1134830f61be] [2023-06-26 20:29:39,267] [WARNING] [koheesio.child2] {__init__.py:<module>:138} - Warning message
[a7d79f7a-f16f-4d2a-8430-1134830f61be] [2023-06-26 20:29:39,267] [ERROR] [koheesio.child2] {__init__.py:<module>:139} - Error message
[a7d79f7a-f16f-4d2a-8430-1134830f61be] [2023-06-26 20:29:39,267] [CRITICAL] [koheesio.child2] {__init__.py:<module>:140} - Critical message
```

--------------------------------

### Initialize LoggingFactory

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/concepts/logger.md

Initialize the LoggingFactory with various configurations: with name, env, and logger_id; with defaults; or with a specific log level.

```python
factory = LoggingFactory(name="replace_koheesio_parent_name", env="local", logger_id="your_run_id")
# Or use default 
factory = LoggingFactory()
# Or just specify log level for koheesio modules
factory = LoggingFactory(level="DEBUG")
```

--------------------------------

### Traditional Spark application for CSV to Delta processing

Source: https://github.com/nike-inc/koheesio/blob/main/docs/tutorials/onboarding.md

A Spark application that reads a CSV file, filters rows where age > 18, calculates average salary per country, and writes the result to a Delta table partitioned by country. This is the baseline example that will be transformed into a Koheesio pipeline.

```python
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, avg

spark = SparkSession.builder.getOrCreate()

# Read data from CSV file
df = spark.read.csv("input.csv", header=True, inferSchema=True)

# Filter data where age is greater than 18
df = df.filter(col("age") > 18)

# Perform aggregation
df = df.groupBy("country").agg(avg("salary").alias("average_salary"))

# Write data to Delta table with partitioning
df.write.format("delta").partitionBy("country").save("/path/to/delta_table")
```

--------------------------------

### Test before pushing with make ci-test-matrix and git push

Source: https://github.com/nike-inc/koheesio/blob/main/docs/development/local-github-actions.md

Catch issues early by testing the most common configuration before pushing with confidence.

```bash
# Test the most common configuration
make ci-test-matrix PYTHON=3.12 PYSPARK=352r

# If that passes, push with confidence
git push
```

--------------------------------

### Run a SparkStep

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/concepts/step.md

Instantiate the custom SparkStep with an input path and call execute() to run it.

```python
step = MySparkStep(input1="path/to/textfile")
step.execute()
```

--------------------------------

### Pre-download Docker image for act

Source: https://github.com/nike-inc/koheesio/blob/main/docs/development/local-github-actions.md

Pre-download the Docker image used by act to speed up the first run.

```bash
docker pull catthehacker/ubuntu:act-latest
```

--------------------------------

### Accessing Multiple Step Outputs

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/concepts/step.md

Shows how to access multiple outputs of a step.

```APIDOC
## Accessing Multiple Step Outputs

### Description
The outputs of a step can be accessed using `self.output.output_name`.

### Method
Property access

### Code Example
```python
step = MyStep(input1="value1", input2=2, input3=3)
step.execute()
print(step.output.output1)  # Outputs: "result"
print(step.output.output2)  # Outputs: 5
```
```

--------------------------------

### Accessing Step Output

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/concepts/step.md

Shows how to access the output of a step using self.output.output_name.

```APIDOC
## Accessing Step Output

### Description
The output of a step can be accessed using `self.output.output_name`.

### Method
Property access

### Code Example
```python
step = MyStep(input1="value1", input2=2)
step.execute()
print(step.output.output1)  # Outputs: "result"
```
```

--------------------------------

### Context Serialization and Deserialization

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/concepts/context.md

Demonstrates loading and saving Context to and from JSON, YAML, and TOML files.

```python
# Load context from a JSON file
context = Context.from_json("path/to/context.json")

# Save context to a JSON file
context.to_json("path/to/context.json")

# Load context from a YAML file
context = Context.from_yaml("path/to/context.yaml")

# Save context to a YAML file
context.to_yaml("path/to/context.yaml")

# Load context from a TOML file
context = Context.from_toml("path/to/context.toml")

# Save context to a TOML file
context.to_toml("path/to/context.toml")
```

--------------------------------

### Create a SparkSession with SparkSession.builder.getOrCreate()

Source: https://github.com/nike-inc/koheesio/blob/main/docs/tutorials/hello-world.md

Creates a SparkSession using the builder pattern. This is required before using Koheesio Spark classes. The session can be local or remote (connect) and can be passed to SparkStep constructors.

```python
from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()
```

--------------------------------

### Context Nested Keys

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/concepts/context.md

Shows how to create a Context with nested keys and access a nested value.

```python
# Create a Context with nested keys
context = Context({
    "database": {
        "bronze_table": "catalog.schema.bronze_table",
        "silver_table": "catalog.schema.silver_table",
        "gold_table": "catalog.schema.gold_table"
    }
})

# Access a nested key
print(context.database.bronze_table)  # Outputs: catalog.schema.bronze_table
```

--------------------------------

### Test minimum and maximum supported versions

Source: https://github.com/nike-inc/koheesio/blob/main/docs/development/local-github-actions.md

Runs the CI test matrix for the minimum and maximum supported Python and PySpark versions to test edge cases.

```bash
# Minimum
make ci-test-matrix PYTHON=3.10 PYSPARK=33

# Maximum  
make ci-test-matrix PYTHON=3.12 PYSPARK=352r
```

--------------------------------

### Partitioning a Delta table with DeltaTableWriter

Source: https://github.com/nike-inc/koheesio/blob/main/docs/tutorials/advanced-data-processing.md

Shows how to configure a Koheesio ETL task to write to a Delta table with partitioning by specified columns. The target is set to a DeltaTableWriter with the table name and partitionBy list.

```python
from koheesio.spark.writers.delta import DeltaTableWriter
from koheesio.spark.etl_task import EtlTask

class MyTask(EtlTask):
    target = DeltaTableWriter(table="my_table", partitionBy=["column1", "column2"])
```

--------------------------------

### Run linting and standards check with make check

Source: https://github.com/nike-inc/koheesio/blob/main/CONTRIBUTING.md

Runs `pylint` and `black` to maintain code standards. Ensure no errors or warnings are reported before submitting a pull request.

```bash
make check
```

--------------------------------

### Define a Step with multiple inputs and outputs

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/concepts/step.md

Extend the Step class with multiple input fields and multiple output fields. The execute method sets each output; here output2 is computed from input2 and input3.

```python
class MyStep(Step):
    input1: str = Field(...)
    input2: int = Field(...)
    input3: int = Field(...)
    
    class Output(StepOutput):
        output1: str = Field(...)
        output2: int = Field(...)
    
    def execute(self):
        # Your logic here
        self.output.output1 = "result"
        self.output.output2 = self.input2 + self.input3
```

--------------------------------

### Serializing a Step

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/concepts/step.md

Shows how to serialize a step to a YAML string using the to_yaml method.

```APIDOC
## Serializing a Step

### Description
You can serialize a step to a YAML string using the `to_yaml` method.

### Method
`to_yaml()`

### Code Example
```python
step = MyStep(input1="value1", input2=2)
yaml_str = step.to_yaml()
```
```

--------------------------------

### Import LoggingFactory

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/concepts/logger.md

Import the LoggingFactory class from the koheesio.logger module.

```python
from koheesio.logger import LoggingFactory
```

--------------------------------

### Transformation.execute

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/spark/transformations.md

The execute method is the main entry point for implementing a transformation. Subclasses should override this method to take a DataFrame from self.input.df, apply a transformation, and store the result in self.output.df.

```APIDOC
## Transformation.execute

### Description
The `execute` method is the core method that subclasses must implement to define a transformation. It should read the input DataFrame from `self.input.df`, apply the transformation logic, and write the result to `self.output.df`.

### Method
`execute`

### Parameters
- `self` - The Transformation instance.

### Returns
- None - The method is expected to set `self.output.df` with the transformed DataFrame.

### Notes
- For `ColumnsTransformation` and `ColumnsTransformationWithTarget`, the `execute` method is already implemented in the base class. Instead, subclasses implement a `func` method that returns the transformation to apply to each column.
```

--------------------------------

### Create Logger with get_logger

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/concepts/logger.md

Create a logger using the get_logger method of LoggingFactory. Optionally inherit from the koheesio logger.

```python
logger = LoggingFactory.get_logger(name=factory.LOGGER_NAME)
# Or for koheesio modules
logger = LoggingFactory.get_logger(name=factory.LOGGER_NAME,inherit_from_koheesio=True)
```

--------------------------------

### Run act with architecture override for M1/M2 Macs

Source: https://github.com/nike-inc/koheesio/blob/main/docs/development/local-github-actions.md

If architecture issues persist on M1/M2 Macs, run act with the linux/amd64 container architecture and specify the matrix versions.

```bash
act -j tests --container-architecture linux/amd64 --matrix python-version:3.12 --matrix pyspark-version:352r
```

--------------------------------

### Context Recursive Merging

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/concepts/context.md

Demonstrates merging two Contexts with nested keys, showing that values are merged recursively and the output of the merged Context.

```python
# Create two Contexts with nested keys
context1 = Context({
    "database": {
        "bronze_table": "catalog.schema.bronze_table",
        "silver_table": "catalog.schema.silver_table"
    }
})

context2 = Context({
    "database": {
        "silver_table": "catalog.schema.new_silver_table",
        "gold_table": "catalog.schema.gold_table"
    }
})

# Merge the two Contexts
context1.merge(context2)

# Print the merged Context
print(context1.to_dict())  
# Outputs: 
# {
#     "database": {
#         "bronze_table": "catalog.schema.bronze_table",
#         "silver_table": "catalog.schema.new_silver_table",
#         "gold_table": "catalog.schema.gold_table"
#     }
# }
```

--------------------------------

### Define a Step with single input and output

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/concepts/step.md

Subclass Step and implement execute. Inputs are declared as class attributes with Field(...), and outputs are declared in a nested Output class that subclasses StepOutput. Access inputs via self.input_name and set outputs via self.output.output_name.

```python
class MyStep(Step):
    input1: str = Field(...)
    input2: int = Field(...)
    
    class Output(StepOutput):
        output1: str = Field(...)
    
    def execute(self):
        # Your logic here
        self.output.output1 = "result"
```

--------------------------------

### Define a Writer subclass with execute method

Source: https://github.com/nike-inc/koheesio/blob/main/docs/reference/spark/writers.md

Shows the minimal structure for a custom Writer. Subclass Writer and implement execute() to read from self.input.df and write to a destination.

```python
class MyWriter(Writer):
    def execute(self):
        # get data from self.input.df
        data = self.input.df
        # write data to destination
        write_to_destination(data)
```

--------------------------------

### Test multiple combinations with a script

Source: https://github.com/nike-inc/koheesio/blob/main/docs/development/local-github-actions.md

A bash script that loops over Python versions 3.10, 3.11, and 3.12, running `make ci-test-matrix` with PySpark 35 for each.

```bash
#!/bin/bash
for python in 3.10 3.11 3.12; do
  echo "Testing Python $python..."
  make ci-test-matrix PYTHON=$python PYSPARK=35
done
```

--------------------------------

### Check syntax and what would run with make ci-dry-run

Source: https://github.com/nike-inc/koheesio/blob/main/docs/development/local-github-actions.md

Use this command to perform a dry run of the CI workflow to check syntax and see what would run without executing.

```bash
# Check syntax and what would run
make ci-dry-run
```
Tokens:
Raw