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