





import os
import posixpath
import uuid
from dataclasses import dataclass
from typing import TYPE_CHECKING, Iterable, List, Optional, Tuple, Union

import numpy as np
import pyarrow as pa

import datasets
from datasets.arrow_writer import ArrowWriter, ParquetWriter
from datasets.config import MAX_SHARD_SIZE
from datasets.filesystems import (
    is_remote_filesystem,
    rename,
)
from datasets.iterable_dataset import _BaseExamplesIterable
from datasets.utils.py_utils import convert_file_size_to_int


__lowerCamelCase	:					int			  =					datasets.utils.logging.get_logger(__name__)

if TYPE_CHECKING:
    import pyspark



@dataclass
class  A__       (	datasets.BuilderConfig      ):
      _UpperCAmelCase     :List[str]						   =       None



def 		A_    (					_lowerCAmelCase							,       _lowerCAmelCase							,       )						->			Any:
 import pyspark

 def generate_fn():
  UpperCamelCase     : str	     =       df.select("*"							,       pyspark.sql.functions.spark_partition_id().alias("part_id"					)					)
  for partition_id in partition_order:
   UpperCamelCase     : Optional[Any]	     =       df_with_partition_id.select("*"					).where(F"""part_id = {partition_id}"""					).drop("part_id"					)
   UpperCamelCase     : Tuple	     =       partition_df.collect()
   UpperCamelCase     : str	     =       0
   for row in rows:
    yield F"""{partition_id}_{row_id}""", row.asDict()
    row_id += 1

 return generate_fn



class  A__       (	_BaseExamplesIterable      ):





      def __init__(			self	,							A_	,							A_=None	,							):
       '''simple docstring'''




       UpperCamelCase     : List[Any]	     =       df
       UpperCamelCase     : int	     =       partition_order or range(self.df.rdd.getNumPartitions()       )
       UpperCamelCase     : int	     =       _generate_iterable_examples(self.df	,							self.partition_order       )





      def __iter__(			self       ):
       '''simple docstring'''




       yield from self.generate_examples_fn()





      def 	__UpperCamelCase(			self	,							A_       ):
       '''simple docstring'''




       UpperCamelCase     : List[str]	     =       list(range(self.df.rdd.getNumPartitions()       )       )
       generator.shuffle(UpperCamelCase__       )
       return SparkExamplesIterable(self.df	,							partition_order=UpperCamelCase__       )





      def 	__UpperCamelCase(			self	,							A_	,							A_       ):
       '''simple docstring'''




       UpperCamelCase     : int	     =       self.split_shard_indices_by_worker(UpperCamelCase__	,							UpperCamelCase__       )
       return SparkExamplesIterable(self.df	,							partition_order=UpperCamelCase__       )





      @property
      def 	__UpperCamelCase(			self       ):
       '''simple docstring'''




       return len(self.partition_order       )



class  A__       (	datasets.DatasetBuilder      ):
      _UpperCAmelCase     :int						   =       SparkConfig





      def __init__(			self	,							A_	,							A_ = None	,							A_ = None	,							**A_	,							):
       '''simple docstring'''




       import pyspark

       UpperCamelCase     : Optional[int]	     =       pyspark.sql.SparkSession.builder.getOrCreate()
       UpperCamelCase     : Union[str, Any]	     =       df
       UpperCamelCase     : str	     =       working_dir

       super().__init__(
           cache_dir=UpperCamelCase__	,							config_name=str(self.df.semanticHash()       )	,							**UpperCamelCase__	,							)





      def 	__UpperCamelCase(			self       ):
       '''simple docstring'''




       def create_cache_and_write_probe(A_       ):
        # makedirs with exist_ok will recursively create the directory. It will not throw an error if directories
        # already exist.
        os.makedirs(self._cache_dir	,							exist_ok=UpperCamelCase__       )
        UpperCamelCase     : List[Any]	     =       os.path.join(self._cache_dir	,							"fs_test" + uuid.uuida().hex       )
        # Opening the file in append mode will create a new file unless it already exists, in which case it will not
        # change the file contents.
        open(UpperCamelCase__	,							"a"       )
        return [probe_file]

       if self._spark.conf.get("spark.master"	,							""       ).startswith("local"       ):
        return

       # If the cluster is multi-node, make sure that the user provided a cache_dir and that it is on an NFS
       # accessible to the driver.
       # TODO: Stream batches to the driver using ArrowCollectSerializer instead of throwing an error.
       if self._cache_dir:
        UpperCamelCase     : Union[str, Any]	     =       (
            self._spark.sparkContext.parallelize(range(1       )	,							1       ).mapPartitions(UpperCamelCase__       ).collect()
        )
        if os.path.isfile(probe[0]       ):
         return

       raise ValueError(
           "When using Dataset.from_spark on a multi-node cluster, the driver and all workers should be able to access cache_dir"       )





      def 	__UpperCamelCase(			self       ):
       '''simple docstring'''




       return datasets.DatasetInfo(features=self.config.features       )





      def 	__UpperCamelCase(			self	,							A_       ):
       '''simple docstring'''




       return [datasets.SplitGenerator(name=datasets.Split.TRAIN       )]





      def 	__UpperCamelCase(			self	,							A_       ):
       '''simple docstring'''




       import pyspark

       def get_arrow_batch_size(A_       ):
        for batch in it:
         yield pa.RecordBatch.from_pydict({"batch_bytes": [batch.nbytes]}       )

       UpperCamelCase     : Optional[Any]	     =       self.df.count()
       UpperCamelCase     : Dict	     =       df_num_rows if df_num_rows <= 100 else 100
       # Approximate the size of each row (in Arrow format) by averaging over a max-100-row sample.
       UpperCamelCase     : Optional[Any]	     =       (
           self.df.limit(UpperCamelCase__       )
           .repartition(1       )
           .mapInArrow(UpperCamelCase__	,							"batch_bytes: long"       )
           .agg(pyspark.sql.functions.sum("batch_bytes"       ).alias("sample_bytes"       )       )
           .collect()[0]
           .sample_bytes
           / sample_num_rows
       )
       UpperCamelCase     : Any	     =       approx_bytes_per_row * df_num_rows
       if approx_total_size > max_shard_size:
        # Make sure there is at least one row per partition.
        UpperCamelCase     : Optional[int]	     =       min(UpperCamelCase__	,							int(approx_total_size / max_shard_size       )       )
        UpperCamelCase     : Tuple	     =       self.df.repartition(UpperCamelCase__       )





      def 	__UpperCamelCase(			self	,							A_	,							A_	,							A_	,							):
       '''simple docstring'''




       import pyspark

       UpperCamelCase     : int	     =       ParquetWriter if file_format == '''parquet''' else ArrowWriter
       UpperCamelCase     : List[Any]	     =       os.path.join(self._working_dir	,							os.path.basename(UpperCamelCase__       )       ) if self._working_dir else fpath
       UpperCamelCase     : Optional[int]	     =       file_format == '''parquet'''

       # Define these so that we don't reference self in write_arrow, which will result in a pickling error due to
       # pickling the SparkContext.
       UpperCamelCase     : Dict	     =       self.config.features
       UpperCamelCase     : int	     =       self._writer_batch_size
       UpperCamelCase     : List[Any]	     =       self._fs.storage_options

       def write_arrow(A_       ):
        # Within the same SparkContext, no two task attempts will share the same attempt ID.
        UpperCamelCase     : Optional[Any]	     =       pyspark.TaskContext().taskAttemptId()
        UpperCamelCase     : Tuple	     =       next(UpperCamelCase__	,							UpperCamelCase__       )
        if first_batch is None:
         # Some partitions might not receive any data.
         return pa.RecordBatch.from_arrays(
             [[task_id], [0], [0]]	,							names=["task_id", "num_examples", "num_bytes"]	,							)
        UpperCamelCase     : List[Any]	     =       0
        UpperCamelCase     : Optional[int]	     =       writer_class(
            features=UpperCamelCase__	,							path=working_fpath.replace("SSSSS"	,							F"""{shard_id:05d}"""       ).replace("TTTTT"	,							F"""{task_id:05d}"""       )	,							writer_batch_size=UpperCamelCase__	,							storage_options=UpperCamelCase__	,							embed_local_files=UpperCamelCase__	,							)
        UpperCamelCase     : Dict	     =       pa.Table.from_batches([first_batch]       )
        writer.write_table(UpperCamelCase__       )
        for batch in it:
         if max_shard_size is not None and writer._num_bytes >= max_shard_size:
          UpperCamelCase     : List[str]	     =       writer.finalize()
          writer.close()
          yield pa.RecordBatch.from_arrays(
              [[task_id], [num_examples], [num_bytes]]	,							names=["task_id", "num_examples", "num_bytes"]	,							)
          shard_id += 1
          UpperCamelCase     : Dict	     =       writer_class(
              features=writer._features	,							path=working_fpath.replace("SSSSS"	,							F"""{shard_id:05d}"""       ).replace("TTTTT"	,							F"""{task_id:05d}"""       )	,							writer_batch_size=UpperCamelCase__	,							storage_options=UpperCamelCase__	,							embed_local_files=UpperCamelCase__	,							)
         UpperCamelCase     : str	     =       pa.Table.from_batches([batch]       )
         writer.write_table(UpperCamelCase__       )

        if writer._num_bytes > 0:
         UpperCamelCase     : List[Any]	     =       writer.finalize()
         writer.close()
         yield pa.RecordBatch.from_arrays(
             [[task_id], [num_examples], [num_bytes]]	,							names=["task_id", "num_examples", "num_bytes"]	,							)

        if working_fpath != fpath:
         for file in os.listdir(os.path.dirname(UpperCamelCase__       )       ):
          UpperCamelCase     : str	     =       os.path.join(os.path.dirname(UpperCamelCase__       )	,							os.path.basename(UpperCamelCase__       )       )
          shutil.move(UpperCamelCase__	,							UpperCamelCase__       )

       UpperCamelCase     : Any	     =       (
           self.df.mapInArrow(UpperCamelCase__	,							"task_id: long, num_examples: long, num_bytes: long"       )
           .groupBy("task_id"       )
           .agg(
               pyspark.sql.functions.sum("num_examples"       ).alias("total_num_examples"       )	,							pyspark.sql.functions.sum("num_bytes"       ).alias("total_num_bytes"       )	,							pyspark.sql.functions.count("num_bytes"       ).alias("num_shards"       )	,							pyspark.sql.functions.collect_list("num_examples"       ).alias("shard_lengths"       )	,							)
           .collect()
       )
       for row in stats:
        yield row.task_id, (row.total_num_examples, row.total_num_bytes, row.num_shards, row.shard_lengths)





      def 	__UpperCamelCase(			self	,							A_	,							A_ = "arrow"	,							A_ = None	,							A_ = None	,							**A_	,							):
       '''simple docstring'''




       self._validate_cache_dir()

       UpperCamelCase     : Dict	     =       convert_file_size_to_int(max_shard_size or MAX_SHARD_SIZE       )
       self._repartition_df_if_needed(UpperCamelCase__       )
       UpperCamelCase     : Optional[Any]	     =       not is_remote_filesystem(self._fs       )
       UpperCamelCase     : Union[str, Any]	     =       os.path.join if is_local else posixpath.join

       UpperCamelCase     : Tuple	     =       '''-TTTTT-SSSSS-of-NNNNN'''
       UpperCamelCase     : List[str]	     =       F"""{self.name}-{split_generator.name}{SUFFIX}.{file_format}"""
       UpperCamelCase     : Union[str, Any]	     =       path_join(self._output_dir	,							UpperCamelCase__       )

       UpperCamelCase     : List[str]	     =       0
       UpperCamelCase     : int	     =       0
       UpperCamelCase     : Dict	     =       0
       UpperCamelCase     : Dict	     =       []
       UpperCamelCase     : Dict	     =       []

       for task_id, content in self._prepare_split_single(UpperCamelCase__	,							UpperCamelCase__	,							UpperCamelCase__       ):
        (
            UpperCamelCase
        )     : Union[str, Any]	     =       content
        if num_bytes > 0:
         total_num_examples += num_examples
         total_num_bytes += num_bytes
         total_shards += num_shards
         task_id_and_num_shards.append((task_id, num_shards)       )
         all_shard_lengths.extend(UpperCamelCase__       )

       UpperCamelCase     : Union[str, Any]	     =       total_num_examples
       UpperCamelCase     : List[Any]	     =       total_num_bytes

       # should rename everything at the end
       logger.debug(F"""Renaming {total_shards} shards."""       )
       if total_shards > 1:
        UpperCamelCase     : List[str]	     =       all_shard_lengths

        # Define fs outside of _rename_shard so that we don't reference self in the function, which will result in a
        # pickling error due to pickling the SparkContext.
        UpperCamelCase     : Union[str, Any]	     =       self._fs

        # use the -SSSSS-of-NNNNN pattern
        def _rename_shard(
            A_	,							A_	,							A_	,							):
         rename(
             UpperCamelCase__	,							fpath.replace("SSSSS"	,							F"""{shard_id:05d}"""       ).replace("TTTTT"	,							F"""{task_id:05d}"""       )	,							fpath.replace("TTTTT-SSSSS"	,							F"""{global_shard_id:05d}"""       ).replace("NNNNN"	,							F"""{total_shards:05d}"""       )	,							)

        UpperCamelCase     : Optional[Any]	     =       []
        UpperCamelCase     : List[str]	     =       0
        for i in range(len(UpperCamelCase__       )       ):
         UpperCamelCase     : str	     =       task_id_and_num_shards[i]
         for shard_id in range(UpperCamelCase__       ):
          args.append([task_id, shard_id, global_shard_id]       )
          global_shard_id += 1
        self._spark.sparkContext.parallelize(UpperCamelCase__	,							len(UpperCamelCase__       )       ).map(lambda A_       : _rename_shard(*UpperCamelCase__       )       ).collect()
       else:
        # don't use any pattern
        UpperCamelCase     : Optional[Any]	     =       0
        UpperCamelCase     : Optional[Any]	     =       task_id_and_num_shards[0][0]
        self._rename(
            fpath.replace("SSSSS"	,							F"""{shard_id:05d}"""       ).replace("TTTTT"	,							F"""{task_id:05d}"""       )	,							fpath.replace(UpperCamelCase__	,							""       )	,							)





      def 	__UpperCamelCase(			self	,							A_	,							):
       '''simple docstring'''




       return SparkExamplesIterable(self.df       )










from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_flax_available,
    is_tf_available,
    is_torch_available,
)


__lowerCamelCase	:					str			  =					{"""configuration_encoder_decoder""": ["""EncoderDecoderConfig"""]}

try:
    if not is_torch_available():
        raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
    pass
else:
    __lowerCamelCase	:					Optional[Any]			  =					["""EncoderDecoderModel"""]

try:
    if not is_tf_available():
        raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
    pass
else:
    __lowerCamelCase	:					Optional[Any]			  =					["""TFEncoderDecoderModel"""]

try:
    if not is_flax_available():
        raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
    pass
else:
    __lowerCamelCase	:					Union[str, Any]			  =					["""FlaxEncoderDecoderModel"""]

if TYPE_CHECKING:
    from .configuration_encoder_decoder import EncoderDecoderConfig

    try:
        if not is_torch_available():
            raise OptionalDependencyNotAvailable()
    except OptionalDependencyNotAvailable:
        pass
    else:
        from .modeling_encoder_decoder import EncoderDecoderModel

    try:
        if not is_tf_available():
            raise OptionalDependencyNotAvailable()
    except OptionalDependencyNotAvailable:
        pass
    else:
        from .modeling_tf_encoder_decoder import TFEncoderDecoderModel

    try:
        if not is_flax_available():
            raise OptionalDependencyNotAvailable()
    except OptionalDependencyNotAvailable:
        pass
    else:
        from .modeling_flax_encoder_decoder import FlaxEncoderDecoderModel

else:
    import sys

    __lowerCamelCase	:					Dict			  =					_LazyModule(__name__, globals()["""__file__"""], _import_structure, module_spec=__spec__)
