In my custom streaming dataset i have tried with

def _load(self):
        if self._preview_mode:
            df = self.spark.read.format("delta").load(self._filepath)
            return df.toPandas()
        else:
            return self.spark.readStream.format("delta").load(self._filepath)

def _save(self, df: DataFrame) -> None:

        if self._preview_mode:
            df = self.spark.readStream.format("delta").load(self._filepath)
            df.writeStream \
            .format("parquet") \
            .option("path", "data/08_reporting/") \
            .option("checkpointLocation", "data/08_reporting/_checkpoints/") \
            .start() \
            .awaitTermination()
            return

        try:
            if