# Path: pemi/data_subject.py
# class SparkDataSubject(DataSubject):
#     def __init__(self, spark, df=None, **kwargs):
#         super().__init__(**kwargs)
#         self.spark = spark
#         self.df = df
# 
#         self.cached_test_df = None
# 
#     def to_pd(self):
#         if self.cached_test_df is not None:
#             return self.cached_test_df
# 
#         converted_df = self.df.toPandas()
#         self.cached_test_df = pd.DataFrame()
#         for column in list(converted_df):
#             self.cached_test_df[column] = converted_df[column].apply(self.schema[column].coerce)
# 
#         return self.cached_test_df
# 
#     def from_pd(self, df, **kwargs):
#         self.df = self.spark.createDataFrame(df)
# 
#     def connect_from(self, other):
#         self.spark = other.spark.builder.getOrCreate()
#         self.df = other.df
#         self.validate_schema()


import pytest
import factory
import pyspark
import pemi
import pemi.testing as pt
from pemi.data_subject import SparkDataSubject
from pemi.fields import *



pytestmark = pytest.mark.spark

class DenormalizeBeersPipe(pemi.Pipe):
    def __init__(self, spark_session, **params):
        super().__init__(**params)

        self.source(
            SparkDataSubject,
            name='sales',
            schema=pemi.Schema(
                beer_id=IntegerField(),
                sold_at=DateField(format='%m/%d/%Y'),
                quantity=IntegerField()
            ),
            spark=spark_session
        )

        self.source(

# Path: pemi/data_subject.py
# class SparkDataSubject(DataSubject):
#     def __init__(self, spark, df=None, **kwargs):
#         super().__init__(**kwargs)
#         self.spark = spark
#         self.df = df
# 
#         self.cached_test_df = None
# 
#     def to_pd(self):
#         if self.cached_test_df is not None:
#             return self.cached_test_df
# 
#         converted_df = self.df.toPandas()
#         self.cached_test_df = pd.DataFrame()
#         for column in list(converted_df):
#             self.cached_test_df[column] = converted_df[column].apply(self.schema[column].coerce)
# 
#         return self.cached_test_df
# 
#     def from_pd(self, df, **kwargs):
#         self.df = self.spark.createDataFrame(df)
# 
#     def connect_from(self, other):
#         self.spark = other.spark.builder.getOrCreate()
#         self.df = other.df
#         self.validate_schema()


# Path: tests/test_spark_job.py
import pytest
import factory
import pyspark
import pemi
import pemi.testing as pt
from pemi.data_subject import SparkDataSubject
from pemi.fields import *




pytestmark = pytest.mark.spark

class DenormalizeBeersPipe(pemi.Pipe):
    def __init__(self, spark_session, **params):
        super().__init__(**params)

        self.source(
            SparkDataSubject,
            name='sales',
            schema=pemi.Schema(
                beer_id=IntegerField(),
                sold_at=DateField(format='%m/%d/%Y'),
                quantity=IntegerField()
            ),
            spark=spark_session
        )

        self.source(

            SparkDataSubject,