### Schema Configuration Examples

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/configuration.md

Demonstrates initialization with inferred schema versus explicit schema files.

```python
# tests/data/tables/data.ndjson exists
# No schema/data.json file exists
reinit_local_metastore(spark, json_tables_dir='tests/data/tables')
# Schema is inferred
```

```json
{
    "type": "struct",
    "fields": [
        {
            "name": "id",
            "type": "long",
            "nullable": false,
            "metadata": {}
        },
        {
            "name": "name",
            "type": "string",
            "nullable": true,
            "metadata": {}
        }
    ]
}
```

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

### Install pysparkdt

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/00_START_HERE.md

Install the library using pip.

```bash
pip install pysparkdt
```

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

### Synchronize development environment

Source: https://github.com/datamole-ai/pysparkdt/blob/main/CONTRIBUTING.md

Use this command to install all project dependencies, including development tools, into the virtual environment.

```bash
# Synchronize virtual environment (installs all dependencies including dev)
uv sync --all-groups
```

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

### Configure Deletion Vectors

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/configuration.md

Examples showing how to enable or disable Delta deletion vectors during metastore initialization.

```python
# Enable deletion vectors (default)
reinit_local_metastore(spark, json_tables_dir='tests/data/tables')

# Explicitly disable deletion vectors
reinit_local_metastore(
    spark,
    json_tables_dir='tests/data/tables',
    deletion_vectors=False
)
```

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

### Databricks Notebook Entrypoint

Source: https://github.com/datamole-ai/pysparkdt/blob/main/README.md

Example of a Databricks notebook that imports processing logic from a package and triggers it using spark and dbutils.

```python
# Databricks notebook source
import sys
from pathlib import Path

MODULE_DIR = Path.cwd().parent
sys.path.append(MODULE_DIR.as_posix())

# COMMAND ----------

from myjobpackage.processing import process_data

# COMMAND ----------

input_table = dbutils.widgets.get('input_table')
output_table = dbutils.widgets.get('output_table')

# COMMAND ----------

process_data(
    spark=spark,
    input_table=input_table,
    output_table=output_table,
)
```

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

### Use TableFactory in Contexts

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/types.md

Examples of passing TableFactory functions to metastore initialization or internal write functions.

```python
reinit_local_metastore(
    spark,
    table_factories={
        'table1': table1_factory,
        'table2': table2_factory,
    }
)
```

```python
_write_table(spark, 'my_table', my_factory)
```

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

### Malformed JSON schema example

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/errors.md

Example of a schema file containing invalid JSON syntax that triggers a JSONDecodeError.

```json
// schema/users.json with syntax error
{
    "type": "struct",
    "fields": [
        { invalid json syntax
    ]
}
```

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

### Load Data from External Sources

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/types.md

Pattern for loading external files like CSVs during test setup.

```python
def create_table_from_csv(spark: SparkSession) -> DataFrame:
    # Load data from a CSV file during test setup
    return spark.read.csv('tests/fixtures/data.csv', header=True)

reinit_local_metastore(
    spark,
    table_factories={'imported_data': create_table_from_csv}
)
```

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

### Execute test with mocked API

Source: https://github.com/datamole-ai/pysparkdt/blob/main/README.md

Example test function utilizing the defined mocks.

```python
def test_process_data(
    spark: SparkSession,
    call_api_mock,
):
  ...
```

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

### Minimal PySparkDT Test Setup

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/examples.md

Configures a SparkSession with a local metastore using a pytest module-scoped fixture.

```python
# tests/test_basic.py
from pytest import fixture
from pysparkdt import spark_base, reinit_local_metastore
from pyspark.sql import SparkSession

@fixture(scope='module')
def spark():
    """Create a SparkSession with local metastore."""
    yield from spark_base('tests/tmp/metastore')

def test_simple_job(spark: SparkSession):
    """Test a simple PySpark job."""
    # Initialize tables from NDJSON files
    reinit_local_metastore(spark, json_tables_dir='tests/data/tables')
    
    # Read input table
    df = spark.read.table('input')
    
    # Process data
    result = df.withColumn('doubled', df.value * 2)
    
    # Write output
    result.write.format('delta').mode('overwrite').saveAsTable('output')
    
    # Verify
    output = spark.read.table('output')
    assert output.count() > 0
```

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

### Triggering ValueError in reinit_local_metastore

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/errors.md

Examples of invalid calls to reinit_local_metastore that trigger the ValueError.

```python
reinit_local_metastore(spark)  # ValueError!
```

```python
reinit_local_metastore(
    spark,
    json_tables_dir='tests/data/tables',
    table_factories={'table': my_factory}
)  # ValueError!
```

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

### Define NDJSON Data File

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/examples.md

Example content for an NDJSON data file used in table loading.

```json
{"user_id": 1, "name": "Alice", "created_at": "2024-01-01T00:00:00"}
{"user_id": 2, "name": "Bob", "created_at": "2024-01-02T00:00:00"}
```

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

### Configure pytest Fixtures for SparkSession

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/overview.md

Example of using a module-scoped SparkSession fixture with an autouse fixture to reset tables between tests.

```python
import pytest
from pysparkdt import spark_base, reinit_local_metastore

@pytest.fixture(scope='module')
def spark():
    yield from spark_base('tests/tmp/metastore')

@pytest.fixture(autouse=True)
def reset_tables(spark):
    """Auto-reset tables before each test"""
    reinit_local_metastore(spark, json_tables_dir='tests/data/tables')
    yield

def test_scenario_1(spark):
    # Fresh tables
    ...

def test_scenario_2(spark):
    # Fresh tables (auto-reset ran)
    ...
```

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

### NDJSON Data Example

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/errors.md

Sample NDJSON data that causes map fields to become None when schema keys are not strings.

```json
// data.ndjson
{"mapping": {"123": "value"}}
```

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

### Defining NDJSON Schema with Non-String Map Keys

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/errors.md

Example of a schema configuration that triggers silent failures when map keys are not strings.

```json
// schema/data.json
{
    "type": "struct",
    "fields": [{
        "name": "mapping",
        "type": {
            "type": "map",
            "keyType": "long",  // Non-string key type
            "valueType": "string",
            "containsNull": false
        },
        "nullable": true,
        "metadata": {}
    }]
}
```

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

### Initialize Metastore and Test Data

Source: https://github.com/datamole-ai/pysparkdt/blob/main/README.md

Use reinit_local_metastore to prepare the metastore and perform DataFrame assertions.

```python
from myjobpackage.processing import process_data
from pysparkdt import reinit_local_metastore
from pyspark.testing import assertDataFrameEqual

def test_process_data(
    spark: SparkSession,
):
    reinit_local_metastore(spark, JSON_TABLES_DIR, deletion_vectors=True)
    
    process_data(
        spark=spark,
        input_table='example_input',
        output_table='output',
    )
    
    output = spark.read.format('delta').table('output')
    expected = spark.read.format('delta').table('expected_output')
    
    assertDataFrameEqual(
        actual=output.select(sorted(output.columns)),
        expected=expected.select(sorted(expected.columns)),
    )
```

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

### Invalid StructType format example

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/errors.md

Example of a schema file that is valid JSON but does not conform to the expected PySpark StructType structure.

```json
{"type": "invalid_type"}  // "invalid_type" is not a valid StructType
```

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

### Load Table with Schema File

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/examples.md

Initialize a local metastore and read a table using an existing schema file.

```python
def test_with_schemas(spark):
    reinit_local_metastore(spark, json_tables_dir='tests/data/tables')
    
    users = spark.read.table('users')
    users.printSchema()
    # Shows: user_id LONG, name STRING, created_at TIMESTAMP
```

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

### Initializing metastore from NDJSON

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/quick-reference.md

Initializes the metastore using a directory of NDJSON files, optionally including schema definitions.

```python
from pysparkdt import reinit_local_metastore

def test_job(spark):
    reinit_local_metastore(spark, json_tables_dir='tests/data/tables')
    # Now test using spark.read.table('tablename')
```

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

### Expected directory structure for NDJSON tables

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/errors.md

Illustrates the required file system layout for successful metastore initialization.

```text
tests/data/tables/
├── users.ndjson          # Required
├── events.ndjson         # Required
└── schema/
    ├── users.json        # Optional
    └── events.json       # Optional
```

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

### Initialize Multiple Tables

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/types.md

Registering multiple factory functions simultaneously within the metastore.

```python
def create_input_table(spark: SparkSession) -> DataFrame:
    return spark.createDataFrame(
        [(1, 'x'), (2, 'y')],
        ['id', 'value']
    )

def create_expected_output(spark: SparkSession) -> DataFrame:
    return spark.createDataFrame(
        [(1, 'X'), (2, 'Y')],
        ['id', 'value_upper']
    )

# Initialize both tables at once
reinit_local_metastore(
    spark,
    table_factories={
        'input': create_input_table,
        'expected_output': create_expected_output,
    }
)
```

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

### Example PySpark StructType JSON Schema

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/quick-reference.md

Represents the expected JSON structure for a PySpark schema definition.

```json
{
    "type": "struct",
    "fields": [
        {
            "name": "id",
            "type": "long",
            "nullable": false,
            "metadata": {}
        },
        {
            "name": "name",
            "type": "string",
            "nullable": true,
            "metadata": {}
        }
    ]
}
```

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

### Initialize Local Metastore with NDJSON

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/api-reference/metastore.md

Re-initializes the metastore using a directory of NDJSON files. Ensure the directory structure matches the expected table names.

```python
def reinit_local_metastore(
    spark: SparkSession,
    json_tables_dir: str | None = None,
    deletion_vectors: bool = True,
    table_factories: dict[str, TableFactory] | None = None,
) -> None
```

```python
from pysparkdt import spark_base, reinit_local_metastore
from pytest import fixture

METASTORE_DIR = 'tests/data/tmp/metastore'
JSON_TABLES_DIR = 'tests/data/tables'

@fixture(scope='module')
def spark():
    yield from spark_base(METASTORE_DIR)

def test_process_data(spark):
    # Initialize metastore from NDJSON files
    # Expects: tests/data/tables/input.ndjson
    # Optional: tests/data/tables/schema/input.json
    reinit_local_metastore(spark, JSON_TABLES_DIR)
    
    # Tables are now available
    df = spark.read.table('input')
    assert df.count() > 0
```

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

### Create and Register an NDJSON Table Factory

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/api-reference/metastore.md

Demonstrates initializing a table factory from an NDJSON file and registering it within the local metastore.

```python
from pysparkdt.metastore import _ndjson_table_factory
from pysparkdt import reinit_local_metastore

# Create factory for a table with optional schema file
factory = _ndjson_table_factory('tests/data/tables/users.ndjson')

# Use in reinit_local_metastore
reinit_local_metastore(
    spark,
    table_factories={'users': factory}
)
```

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

### Test Streaming Queries

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/quick-reference.md

Pattern for initializing a local metastore, executing a Delta stream with a checkpoint, and verifying output.

```python
def test_stream(spark):
    reinit_local_metastore(spark, json_tables_dir='tests/data/tables')
    
    # Start stream
    query = (
        spark.readStream.format('delta').table('input')
        .writeStream.format('delta')
        .option('checkpointLocation', 'tests/tmp/_cp')
        .option('path', 'tests/tmp/output')
        .trigger(availableNow=True)
        .start()
    )
    
    # Wait
    query.awaitTermination(60)
    
    # Verify
    output = spark.read.table('output')
```

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

### Map NDJSON directory to TableFactory callables

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/api-reference/metastore.md

Demonstrates initializing a local metastore from a directory of NDJSON files. Requires a directory structure where schema files are located in a subfolder named 'schema'.

```python
# Example directory structure:
# tests/data/tables/
#   ├── input.ndjson
#   ├── expected_output.ndjson
#   └── schema/
#       ├── input.json
#       └── expected_output.json

from pysparkdt.metastore import _ndjson_dir_to_tables
from pysparkdt import reinit_local_metastore

tables = _ndjson_dir_to_tables('tests/data/tables')
# tables = {'input': <factory>, 'expected_output': <factory>}

reinit_local_metastore(spark, table_factories=tables)
```

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

### Initializing metastore from table factories

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/quick-reference.md

Uses programmatic table factories to define tables dynamically during initialization.

```python
def create_users(spark):
    return spark.createDataFrame(
        [(1, 'Alice'), (2, 'Bob')],
        ['id', 'name']
    )

reinit_local_metastore(
    spark,
    table_factories={'users': create_users}
)
```

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

### Configure Deletion Vectors in Tests

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/examples.md

Illustrates toggling deletion vector support during metastore initialization.

```python
def test_with_deletion_vectors(spark):
    """Test with deletion vectors enabled (default)."""
    reinit_local_metastore(
        spark,
        json_tables_dir='tests/data/tables',
        deletion_vectors=True
    )
    
    df = spark.read.table('input')
    df.write.format('delta').mode('overwrite').saveAsTable('temp')
    
    # Delete some rows
    spark.sql("DELETE FROM temp WHERE id > 1")
    
    result = spark.read.table('temp')
    assert result.count() == 1

def test_without_deletion_vectors(spark):
    """Test with deletion vectors disabled."""
    reinit_local_metastore(
        spark,
        json_tables_dir='tests/data/tables',
        deletion_vectors=False
    )
    
    df = spark.read.table('input')
    # Table created without deletion vectors
```

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

### Define Test Constants

Source: https://github.com/datamole-ai/pysparkdt/blob/main/README.md

Set up directory paths for test data and the temporary metastore.

```python
DATA_DIR = f'{os.path.dirname(__file__)}/data'
JSON_TABLES_DIR = f'{DATA_DIR}/tables'
TMP_DIR = f'{DATA_DIR}/tmp'
METASTORE_DIR = f'{TMP_DIR}/metastore'
```

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

### Define MapType with non-string keys

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/overview.md

Example of an invalid schema definition where non-string map keys cause data loading failures.

```python
# Schema with non-string map keys
schema = StructType([
    StructField('mapping', MapType(
        LongType(),      # ❌ Non-string key type
        StringType()
    ), nullable=True)
])

# NDJSON data has string keys (only option in JSON)
# {"mapping": {"123": "value"}}

# Result: mapping field is None instead of {"123": "value"}
```

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

### Handle ValueError in Metastore Initialization

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/examples.md

Demonstrates catching ValueError when parameters for reinit_local_metastore are missing or conflicting.

```python
from pysparkdt import reinit_local_metastore

def test_missing_source():
    """Handle missing data source parameter."""
    try:
        reinit_local_metastore(None)  # Neither parameter provided
    except ValueError as e:
        print(f"Error: {e}")
        # Output: Error: Exactly one of json_tables_dir or table_factories must be provided

def test_conflicting_sources():
    """Handle conflicting parameters."""
    try:
        reinit_local_metastore(
            None,
            json_tables_dir='path',
            table_factories={'table': lambda s: s.createDataFrame([])}
        )
    except ValueError as e:
        print(f"Error: {e}")
        # Output: Error: Exactly one of json_tables_dir or table_factories must be provided
```

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

### Define Test Data and Schema

Source: https://github.com/datamole-ai/pysparkdt/blob/main/README.md

Examples of input data in NDJSON format and the corresponding PySpark StructType schema in JSON format.

```json lines
{"id": 0, "time_utc": "2024-01-08T11:00:00", "name": "Jorge", "feature": 0.5876}
{"id": 1, "time_utc": "2024-01-11T14:28:00", "name": "Ricardo", "feature": 0.42}
```

```json
{
    "type": "struct",
    "fields": 
    [
        {
            "name": "id",
            "type": "long",
            "nullable": false,
            "metadata": {}
        },
        {
            "name": "time_utc",
            "type": "timestamp",
            "nullable": false,
            "metadata": {}
        },
        {
            "name": "name",
            "type": "string",
            "nullable": true,
            "metadata": {}
        },
        {
            "name": "feature",
            "type": "double",
            "nullable": true,
            "metadata": {}
        }
    ]
}
```

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

### Initialize from NDJSON Directory

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/configuration.md

Load tables from a directory containing NDJSON files.

```python
reinit_local_metastore(
    spark,
    json_tables_dir='tests/data/tables'
)
```

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

### Define processing logic with external API calls

Source: https://github.com/datamole-ai/pysparkdt/blob/main/README.md

Example structure for a PySpark job that performs API calls within a partition-level function.

```python
def call_api(
    data_df: pyspark.sql.DataFrame,
) -> pyspark.sql.DataFrame:
    # Call API in parallel (session per partition)
    result = data_df.rdd.mapPartitions(_partition_run).toDF()
    return result
  
def _partition_run(
    iterable: Iterable[Row],
) -> Iterable[dict[str, Any]]:
  with ApiSessionClient() as client:
      for row in iterable:
          ...
          output = client.post(prepared_data)
          ...
          yield output
        
def process_data(
    data_df: pyspark.sql.DataFrame,
) -> pyspark.sql.DataFrame:
    ...
    ... = call_api(...)
    ...
```

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

### Load Table with Schema Inference

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/examples.md

Initialize a local metastore and read a table without a schema file to trigger automatic type inference.

```python
# tests/data/tables/products.ndjson only
{"product_id": "P001", "price": 19.99}
{"product_id": "P002", "price": 29.99}

# Test code
def test_inferred_schema(spark):
    reinit_local_metastore(spark, json_tables_dir='tests/data/tables')
    
    products = spark.read.table('products')
    products.printSchema()
    # Shows: product_id STRING, price DOUBLE
    # (inferred from data)
```

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

### spark_base(metastore_dir: str)

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/quick-reference.md

Initializes a SparkSession with pre-configured settings including Hive support, Delta extensions, and UTC timezone.

```APIDOC
## spark_base(metastore_dir: str)

### Description
Initializes a SparkSession with fixed configurations including Hive support, Delta extensions, and UTC timezone.

### Parameters
- **metastore_dir** (str) - Required - Path for local metastore storage

### Returns
- **Iterator[SparkSession]** - An iterator yielding the configured SparkSession.
```

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

### Testing streaming jobs with await

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/quick-reference.md

Demonstrates how to handle streaming queries in tests by awaiting termination.

```python
def test_streaming_job(spark):
    reinit_local_metastore(spark, json_tables_dir='tests/data/tables')
    
    # Start stream
    query = spark.readStream.format('delta').table('input') \
        .writeStream.format('delta') \
        .option('checkpointLocation', 'tests/tmp/_checkpoint') \
        .start()
    
    # Wait for completion
    query.awaitTermination(timeout=60)
    
    # Verify
    output = spark.read.table('output')
```

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

### Initialize from Table Factories

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/configuration.md

Create tables programmatically using factory functions that return a DataFrame.

```python
def create_users(spark: SparkSession) -> DataFrame:
    return spark.createDataFrame(
        [(1, 'Alice'), (2, 'Bob')],
        ['id', 'name']
    )

reinit_local_metastore(
    spark,
    table_factories={'users': create_users}
)
```

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

### Initialize Local Metastore with Table Factories

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/api-reference/metastore.md

Re-initializes the metastore using a dictionary of TableFactory callables. This is mutually exclusive with json_tables_dir.

```python
from pysparkdt import spark_base, reinit_local_metastore
from pyspark.sql import SparkSession, DataFrame

def create_input_table(spark: SparkSession) -> DataFrame:
    return spark.createDataFrame(
        [
            (1, 'Alice', 25),
            (2, 'Bob', 30),
        ],
        ['id', 'name', 'age']
    )

def create_expected_output(spark: SparkSession) -> DataFrame:
    return spark.createDataFrame(
        [
            (1, 'Alice', 25),
            (2, 'Bob', 30),
        ],
        ['id', 'name', 'age']
    )

def test_job(spark):
    # Initialize metastore programmatically
    reinit_local_metastore(
        spark,
        table_factories={
            'input': create_input_table,
            'expected_output': create_expected_output,
        }
    )
    
    # Process the input table
    df = spark.read.table('input')
    result = df.filter(df.age > 25)
    result.write.format('delta').mode('overwrite').saveAsTable('output')
    
    # Verify results
    output = spark.read.table('output')
    expected = spark.read.table('expected_output')
```

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

### Streaming Job Testing

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/overview.md

Demonstrates testing Spark streaming jobs by initializing the metastore and awaiting query termination.

```python
def test_streaming_job(spark):
    reinit_local_metastore(spark, json_tables_dir='tests/data/tables')
    
    # Start streaming job
    query = process_data_streaming(
        spark,
        input_table='input',
        output_table='output',
        checkpoint_location='tests/tmp/_checkpoint'
    )
    
    # Wait for completion
    query.awaitTermination(timeout=60)
    
    # Verify results
    output = spark.read.table('output')
```

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

### Simple PySpark Job Testing

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/examples.md

Demonstrates a basic job processing function and its corresponding test using a local metastore.

```python
# myjobpackage/processing.py
from pyspark.sql import SparkSession

def process_data(
    spark: SparkSession,
    input_table: str,
    output_table: str
) -> None:
    """Process input table and save to output."""
    df = spark.read.table(input_table)
    
    result = (
        df
        .filter(df.value > 0)
        .withColumn('processed', df.value * 100)
    )
    
    result.write.format('delta').mode('overwrite').saveAsTable(output_table)

# tests/test_processing.py
from pyspark.testing import assertDataFrameEqual
from myjobpackage.processing import process_data

def test_process_data(spark):
    reinit_local_metastore(spark, json_tables_dir='tests/data/tables')
    
    # Run job
    process_data(
        spark,
        input_table='input',
        output_table='output'
    )
    
    # Verify
    output = spark.read.table('output')
    expected = spark.read.table('expected_output')
    
    assertDataFrameEqual(
        actual=output.select(sorted(output.columns)),
        expected=expected.select(sorted(expected.columns))
    )
```

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

### reinit_local_metastore(spark, json_tables_dir=None, deletion_vectors=True, table_factories=None)

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/configuration.md

Initializes the local metastore by dropping existing tables and creating new ones based on the provided source (NDJSON directory or table factories).

```APIDOC
## reinit_local_metastore

### Description
Resets the local Spark metastore and populates it with tables. This function drops all existing tables before creating new ones from the specified source.

### Signature
`reinit_local_metastore(spark: SparkSession, json_tables_dir: str | None = None, deletion_vectors: bool = True, table_factories: dict[str, TableFactory] | None = None) -> None`

### Parameters
- **spark** (SparkSession) - Required - The Spark session to initialize tables in.
- **json_tables_dir** (str | None) - Optional - Directory containing NDJSON files and optional schema subdirectory. Mutually exclusive with `table_factories`.
- **deletion_vectors** (bool) - Optional - Whether to enable Delta deletion vectors on created tables. Defaults to True.
- **table_factories** (dict[str, TableFactory] | None) - Optional - Programmatically-defined table factories. Mutually exclusive with `json_tables_dir`.

### Usage Examples

#### NDJSON Directory Mode
```python
reinit_local_metastore(spark, json_tables_dir='tests/data/tables')
```

#### Table Factories Mode
```python
def create_users(spark: SparkSession) -> DataFrame:
    return spark.createDataFrame([(1, 'Alice'), (2, 'Bob')], ['id', 'name'])

reinit_local_metastore(spark, table_factories={'users': create_users})
```
```

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

### Import Public API Components

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/README.md

Import the core utilities required for initializing Spark sessions and managing the local metastore.

```python
from pysparkdt import spark_base, reinit_local_metastore, TableFactory
```

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

### Check Metastore and Table Locations

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/quick-reference.md

Commands to verify active databases and inspect specific table storage paths.

```python
# Verify metastore directory is being used
spark.sql('SHOW DATABASES').show()

# Check table locations
spark.sql('DESCRIBE FORMATTED table_name').show()
```

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

### Standard Test Directory Structure

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/overview.md

Organize test files and data assets using a consistent directory layout for easier maintenance.

```text
tests/
├── test_processing.py
├── test_streaming.py
└── data/
    ├── tables/
    │   ├── input.ndjson
    │   ├── expected_output.ndjson
    │   └── schema/
    │       ├── input.json
    │       └── expected_output.json
    └── tmp/
        └── metastore/  (generated)
```

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

### Programmatic Table Initialization

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/overview.md

Initializes tables using factory callables instead of loading from JSON files.

```python
def test_job_with_factories(spark):
    reinit_local_metastore(
        spark,
        table_factories={
            'input': lambda spark: spark.createDataFrame([...]),
            'expected': lambda spark: spark.createDataFrame([...]),
        }
    )
    # Test execution
```

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

### Handle reinit_local_metastore parameters

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/quick-reference.md

Ensure exactly one of json_tables_dir or table_factories is provided to the reinit_local_metastore function.

```python
# ❌ Wrong: neither parameter
reinit_local_metastore(spark)

# ❌ Wrong: both parameters
reinit_local_metastore(spark, json_tables_dir='...', table_factories={...})

# ✓ Correct
reinit_local_metastore(spark, json_tables_dir='...')
reinit_local_metastore(spark, table_factories={...})
```

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

### Metastore Reset Behavior

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/configuration.md

Demonstrates how subsequent calls to reinit_local_metastore replace the existing table set.

```python
def test_scenario_1(spark):
    reinit_local_metastore(spark, json_tables_dir='tests/data/tables')
    # Table 'users' exists
    
    # Calling again resets all tables
    reinit_local_metastore(spark, table_factories={'products': create_products})
    # Table 'users' is gone, only 'products' exists
```

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

### Create Simple DataFrame

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/types.md

A basic implementation of a TableFactory using a list of tuples and a schema list.

```python
def create_users_table(spark: SparkSession) -> DataFrame:
    data = [
        (1, 'Alice', 25),
        (2, 'Bob', 30),
        (3, 'Charlie', 35),
    ]
    schema = ['id', 'name', 'age']
    return spark.createDataFrame(data, schema)

# Use with reinit_local_metastore
reinit_local_metastore(
    spark,
    table_factories={'users': create_users_table}
)
```

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

### Testing Multiple Tables with Join

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/examples.md

Initializes multiple tables using factory functions to test join operations within a job.

```python
def test_multi_table_join(spark):
    """Test job that joins multiple tables."""
    reinit_local_metastore(
        spark,
        table_factories={
            'users': lambda s: s.createDataFrame(
                [(1, 'Alice'), (2, 'Bob')],
                ['user_id', 'name']
            ),
            'orders': lambda s: s.createDataFrame(
                [(1, 1, 100.0), (2, 1, 200.0)],
                ['order_id', 'user_id', 'amount']
            ),
        }
    )
    
    # Test join
    users = spark.read.table('users')
    orders = spark.read.table('orders')
    
    result = users.join(orders, 'user_id')
    result.write.format('delta').mode('overwrite').saveAsTable('output')
    
    output = spark.read.table('output')
    assert output.count() == 2
```

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

### View Documentation Directory Structure

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/README.md

Displays the file hierarchy of the documentation directory.

```text
/output/
├── README.md                    (this file)
├── INDEX.md                     Navigation and cross-references
├── overview.md                  Project overview and concepts
├── quick-reference.md           Common tasks and snippets
├── examples.md                  Detailed usage examples
├── types.md                     Type definitions
├── configuration.md             Configuration guide
├── errors.md                    Error reference
└── api-reference/
    ├── spark_base.md            SparkSession factory
    └── metastore.md             Metastore management
```

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

### Catching JSONDecodeError

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/errors.md

Demonstrates how to wrap the metastore initialization in a try-except block to handle schema parsing errors.

```python
import json
from pysparkdt import reinit_local_metastore

try:
    reinit_local_metastore(spark, json_tables_dir='tests/data/tables')
except json.JSONDecodeError as e:
    print(f"Invalid schema file: {e}")
```

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

### Test UDFs with Null Values

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/examples.md

Demonstrates testing a UDF that handles null values using a table factory to initialize the metastore.

```python
from pyspark.sql import functions as F
from pyspark.sql.types import StringType

def create_data_with_nulls(spark):
    """Create data including null values."""
    return spark.createDataFrame(
        [(1, 'Alice'), (2, None), (3, 'Bob')],
        ['id', 'name']
    )

def test_udf_with_nulls(spark):
    """Test UDF behavior with null values."""
    reinit_local_metastore(
        spark,
        table_factories={'input': create_data_with_nulls}
    )
    
    # Define UDF
    @F.udf(StringType())
    def process_name(name):
        return name.upper() if name else 'UNKNOWN'
    
    df = spark.read.table('input')
    result = df.withColumn('name_upper', process_name(df.name))
    
    result.write.format('delta').mode('overwrite').saveAsTable('output')
    output = spark.read.table('output')
    
    assert output.count() == 3
```

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

### Testing with Deletion Vectors Disabled

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/configuration.md

Demonstrates how to explicitly disable deletion vectors during metastore initialization.

```python
def test_no_deletion_vectors(spark):
    reinit_local_metastore(
        spark,
        json_tables_dir='tests/data/tables',
        deletion_vectors=False  # Explicitly disable
    )
    # Test with deletion vectors off
    ...
```

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

### Resolving the ValueError

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/errors.md

Correct usage patterns for reinit_local_metastore by providing exactly one source.

```python
reinit_local_metastore(spark, json_tables_dir='tests/data/tables')
```

```python
def create_table(spark):
    return spark.createDataFrame([...])

reinit_local_metastore(
    spark,
    table_factories={'my_table': create_table}
)
```

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

### Create Programmatic Tables with Factories

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/examples.md

Use factory functions to define and register tables programmatically for testing.

```python
from pyspark.sql import SparkSession, DataFrame
from pyspark.sql.types import (
    StructType, StructField,
    LongType, StringType, TimestampType
)

# Define schema once
USER_SCHEMA = StructType([
    StructField('user_id', LongType(), nullable=False),
    StructField('name', StringType(), nullable=True),
    StructField('email', StringType(), nullable=True),
])

# Create factory function
def create_users_table(spark: SparkSession) -> DataFrame:
    """Create sample users table."""
    data = [
        (1, 'Alice', 'alice@example.com'),
        (2, 'Bob', 'bob@example.com'),
        (3, 'Charlie', 'charlie@example.com'),
    ]
    return spark.createDataFrame(data, USER_SCHEMA)

# Use in test
def test_with_factories(spark):
    reinit_local_metastore(
        spark,
        table_factories={'users': create_users_table}
    )
    
    users = spark.read.table('users')
    assert users.count() == 3
```

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

### Testing with Column Ordering

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/examples.md

Ensures output verification is robust against column order variations by sorting columns before comparison.

```python
def test_output_columns(spark):
    """Verify output has correct columns regardless of order."""
    reinit_local_metastore(spark, json_tables_dir='tests/data/tables')
    
    # Run job
    df = spark.read.table('input')
    result = df.select('name', 'value', 'id')
    result.write.format('delta').mode('overwrite').saveAsTable('output')
    
    # Sort columns for comparison
    output = spark.read.table('output')
    expected = spark.read.table('expected_output')
    
    output_sorted = output.select(sorted(output.columns))
    expected_sorted = expected.select(sorted(expected.columns))
    
    assertDataFrameEqual(output_sorted, expected_sorted)
```

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

### Batch Testing with Fixed Data

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/configuration.md

Uses a module-scoped spark fixture and an autouse fixture to reset the metastore before each test.

```python
@fixture(scope='module')
def spark():
    yield from spark_base('tests/tmp/metastore')

@pytest.fixture(autouse=True)
def reset_metastore(spark):
    """Reset metastore before each test"""
    reinit_local_metastore(spark, json_tables_dir='tests/data/tables')
    yield

def test_scenario_1(spark):
    # Tables are fresh
    ...

def test_scenario_2(spark):
    # Tables are fresh (reset happened before this test)
    ...
```

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

### Handle FileNotFoundError in Metastore Initialization

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/examples.md

Catches FileNotFoundError when the provided JSON tables directory does not exist.

```python
from pysparkdt import reinit_local_metastore

def test_missing_directory(spark):
    """Handle missing data directory."""
    try:
        reinit_local_metastore(spark, json_tables_dir='/nonexistent/path')
    except FileNotFoundError as e:
        print(f"Data directory not found: {e}")
        # Take corrective action
```

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

### Organize project file structure

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/quick-reference.md

Recommended directory layout for PySparkDT projects, separating source code, tests, and data artifacts.

```text
myproject/
├── src/
│   └── mypackage/
│       ├── __init__.py
│       └── processing.py       # Core logic
├── tests/
│   ├── conftest.py             # pytest configuration
│   ├── test_processing.py      # Tests
│   └── data/
│       ├── tables/
│       │   ├── input.ndjson
│       │   ├── expected.ndjson
│       │   └── schema/
│       │       ├── input.json
│       │       └── expected.json
│       └── tmp/
│           └── metastore/      # Generated during tests
└── pyproject.toml
```

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

### spark_base

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/MANIFEST.md

Initializes the Spark session for the application. This is the primary entry point for setting up the Spark environment.

```APIDOC
## spark_base

### Description
Initializes and configures the Spark session for use within the application.

### Parameters
- Refer to the official library documentation for specific signature and parameter details.

### Exceptions
- Raises standard Spark initialization exceptions if configuration fails.
```

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

### Check code formatting and linting

Source: https://github.com/datamole-ai/pysparkdt/blob/main/CONTRIBUTING.md

Commands to verify that code adheres to formatting and linting standards without applying changes.

```bash
uv run ruff format --check .
uv run ruff check .
```

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

### Basic Testing Pattern

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/overview.md

Uses a pytest fixture to manage the SparkSession lifecycle and reinitializes the metastore before executing test logic.

```python
from pytest import fixture
from pysparkdt import spark_base, reinit_local_metastore

@fixture(scope='module')
def spark():
    yield from spark_base('tests/tmp/metastore')

def test_job(spark):
    reinit_local_metastore(spark, json_tables_dir='tests/data/tables')
    
    # Job code here
    df = spark.read.table('input')
    result = process_data(df)
    result.write.format('delta').mode('overwrite').saveAsTable('output')
    
    # Assertions
    expected = spark.read.table('expected_output')
    assert result.collect() == expected.collect()
```

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

### Catching the ValueError

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/errors.md

Demonstrates how to handle the ValueError using a try-except block.

```python
from pysparkdt import reinit_local_metastore

try:
    # This will raise ValueError
    reinit_local_metastore(spark)
except ValueError as e:
    print(f"Configuration error: {e}")
```

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

### RDD and UDF Mocking

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/overview.md

Shows how to mock external dependencies using pytest-mock when testing code that utilizes RDD mapPartitions or UDFs.

```python
@fixture
def mock_api_client(mocker):
    # Mock the API client before test runs
    mocker.patch.object(
        my_package.processing,
        'ApiSessionClient',
        return_value=mock_client
    )

def test_with_rdd_map(spark, mock_api_client):
    reinit_local_metastore(spark, json_tables_dir='tests/data/tables')
    # Mocked API client is used throughout
```

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

### Catching PySpark Exceptions

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/errors.md

Demonstrates how to catch specific PySpark and standard Python exceptions during the reinit_local_metastore process.

```python
from pyspark.errors import AnalysisException, PySparkException
from pysparkdt import reinit_local_metastore

try:
    reinit_local_metastore(spark, json_tables_dir='tests/data/tables')
except (ValueError, FileNotFoundError, AnalysisException, PySparkException) as e:
    print(f"Failed to initialize metastore: {e}")
```

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

### Test Basic Streaming Job

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/examples.md

Tests a streaming job that reads from and writes to Delta tables. Requires a pre-initialized local metastore.

```python
from pyspark.sql import SparkSession

def test_streaming_job(spark: SparkSession):
    """Test streaming job that reads and writes Delta."""
    reinit_local_metastore(spark, json_tables_dir='tests/data/tables')
    
    # Start streaming query
    query = (
        spark.readStream.format('delta')
        .table('input')
        .writeStream.format('delta')
        .option('checkpointLocation', 'tests/tmp/_checkpoint')
        .mode('append')
        .start()
    )
    
    # Wait for completion
    query.awaitTermination(timeout=60)
    
    # Verify results
    output = spark.read.table('output')
    expected = spark.read.table('expected_output')
    
    assert output.count() == expected.count()
```

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

### spark_base(metastore_dir: str) -> Iterator[SparkSession]

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/api-reference/spark_base.md

Creates and yields a SparkSession configured for local testing with a dynamically created local metastore acting as the Databricks data catalog.

```APIDOC
## spark_base(metastore_dir: str)

### Description
Creates and yields a SparkSession configured for local testing with a dynamically created local metastore. It handles session lifecycle, including teardown of existing sessions and cleanup of the metastore directory.

### Parameters
- **metastore_dir** (str) - Required - The directory path to use for the dynamically created local metastore.

### Return Type
- **Iterator[SparkSession]** - Yields a SparkSession configured with Hive support, Delta Lake extension, UTC timezone, and error-level logging.

### Usage Example
```python
from pytest import fixture
from pysparkdt import spark_base

METASTORE_DIR = 'tests/data/tmp/metastore'

@fixture(scope='module')
def spark():
    yield from spark_base(METASTORE_DIR)
```
```

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

### Inspect Available Tables

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/quick-reference.md

Methods to list tables in the current database or read a specific table into a DataFrame.

```python
# List all tables in current database
tables_df = spark.sql('SHOW TABLES')
tables_df.show()

# Or as list
table_names = [row.tableName for row in spark.sql('SHOW TABLES').collect()]
print(table_names)

# Read a specific table
df = spark.read.table('table_name')
df.show()
```

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

### Inspect Available Tables in PySpark

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/examples.md

Lists all tables currently registered in the metastore using the provided directory.

```python
def test_debug_tables(spark):
    """Inspect what tables are available."""
    reinit_local_metastore(spark, json_tables_dir='tests/data/tables')
    
    # Show available tables
    tables = spark.sql('SHOW TABLES')
    tables.show()
    
    # List table names
    table_names = [row.tableName for row in tables.collect()]
    print(f"Tables: {table_names}")
```

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

### Configure Spark Fixture

Source: https://github.com/datamole-ai/pysparkdt/blob/main/README.md

Create a pytest fixture for a local Spark session using the spark_base function.

```python
from pytest import fixture
from pysparkdt import spark_base

@fixture(scope='module')
def spark():
    yield from spark_base(METASTORE_DIR)
```

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

### Minimal PySpark Test

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/00_START_HERE.md

A basic pytest fixture and test case demonstrating how to initialize a local metastore and read a table.

```python
from pytest import fixture
from pysparkdt import spark_base, reinit_local_metastore

@fixture(scope='module')
def spark():
    yield from spark_base('tests/tmp/metastore')

def test_job(spark):
    reinit_local_metastore(spark, json_tables_dir='tests/data/tables')
    df = spark.read.table('input')
    assert df.count() > 0
```

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

### Function Signature

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/configuration.md

The definition of the reinit_local_metastore function and its parameters.

```python
def reinit_local_metastore(
    spark: SparkSession,
    json_tables_dir: str | None = None,
    deletion_vectors: bool = True,
    table_factories: dict[str, TableFactory] | None = None,
) -> None
```

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

### Configure MapType keys

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/quick-reference.md

Map keys must be of StringType to avoid silent failures where non-string keys result in None values.

```python
# ❌ Wrong: non-string map keys
MapType(LongType(), StringType())

# ✓ Correct: string keys
MapType(StringType(), StringType())
```

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

### Create a custom SparkSession fixture

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/configuration.md

A template for defining a custom SparkSession fixture using pytest, allowing for configuration overrides and additions.

```python
from delta import configure_spark_with_delta_pip
from pyspark.sql import SparkSession
from pytest import fixture

@fixture(scope='module')
def custom_spark():
    builder = (
        SparkSession.builder
        .appName('my_test_app')
        .config('spark.sql.warehouse.dir', 'my_warehouse')
        .config('spark.driver.extraJavaOptions', '-Dderby.system.home=my_warehouse')
        .enableHiveSupport()
        .config('spark.sql.extensions', 'io.delta.sql.DeltaSparkSessionExtension')
        .config('spark.sql.catalog.spark_catalog', 'org.apache.spark.sql.delta.catalog.DeltaCatalog')
        .config('spark.sql.session.timeZone', 'UTC')
        .config('spark.sql.adaptive.enabled', 'true')  # Additional configs here
    )
    
    session = configure_spark_with_delta_pip(builder).getOrCreate()
    session.sparkContext.setLogLevel('ERROR')
    
    yield session
    
    # Cleanup
    session.stop()
```

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

### Define Tables with Table Factories

Source: https://github.com/datamole-ai/pysparkdt/blob/main/README.md

Use a dictionary mapping table names to callables that return a DataFrame to define tables programmatically.

```python
from pyspark.sql import SparkSession
from pysparkdt import reinit_local_metastore

def _input(spark: SparkSession):
    return spark.createDataFrame(...)

def _expected_output(spark: SparkSession):
    return spark.createDataFrame(...)

TABLE_FACTORIES = {
    'example_input': _input,
    'expected_output': _expected_output,
}

reinit_local_metastore(spark, table_factories=TABLE_FACTORIES)
```

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

### Format and lint code

Source: https://github.com/datamole-ai/pysparkdt/blob/main/CONTRIBUTING.md

Commands to automatically format code and apply fixes for linting errors.

```bash
uv run ruff format .
uv run ruff check . --fix
```

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

### Test Complex Nested Schemas

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/examples.md

Shows how to define and test tables containing ArrayType and MapType structures.

```python
from pyspark.sql.types import (
    StructType, StructField,
    MapType, ArrayType,
    LongType, StringType, DoubleType
)

def create_complex_table(spark):
    """Create table with complex nested types."""
    schema = StructType([
        StructField('id', LongType(), nullable=False),
        StructField('tags', ArrayType(StringType()), nullable=True),
        StructField('metadata', MapType(StringType(), StringType()), nullable=True),
    ])
    
    data = [
        (1, ['tag1', 'tag2'], {'key1': 'value1'}),
        (2, ['tag3'], {'key2': 'value2'}),
    ]
    
    return spark.createDataFrame(data, schema)

def test_complex_schema(spark):
    """Test processing of complex nested structures."""
    reinit_local_metastore(
        spark,
        table_factories={'input': create_complex_table}
    )
    
    df = spark.read.table('input')
    
    # Explode array
    exploded = df.select('id', F.explode('tags').alias('tag'))
    assert exploded.count() == 3
```

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

### Import TableFactory

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/types.md

The standard import statement for the TableFactory type.

```python
from pysparkdt import TableFactory
```

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

### Implement TableFactory Function

Source: https://github.com/datamole-ai/pysparkdt/blob/main/_autodocs/types.md

The expected signature for a function that creates and returns a DataFrame.

```python
def my_table_factory(spark: SparkSession) -> DataFrame:
    # Create and return a PySpark DataFrame
    return spark.createDataFrame(...)
```