



"""simple docstring"""







import copy
import os
import tempfile
from unittest import TestCase
from unittest.mock import patch

import numpy as np
import pyarrow as pa
import pyarrow.parquet as pq
import pytest

from datasets.arrow_writer import ArrowWriter, OptimizedTypedSequence, ParquetWriter, TypedSequence
from datasets.features import ArrayaD, ClassLabel, Features, Image, Value
from datasets.features.features import ArrayaDExtensionType, cast_to_python_objects
from datasets.keyhash import DuplicatedKeysError, InvalidKeyError

from .utils import require_pil
class        __A       (	A_    ):







		'''simple docstring'''





		def 					UpperCAmelCase    (	self						:  Dict						)					->		List[str]:






				"""simple docstring"""



				lowercase__   :    int				=			pa.array(TypedSequence([1, 2, 3]						)						)
				self.assertEqual(arr.type			,pa.intaa()						)







		def 					UpperCAmelCase    (	self						:  Optional[Any]						)					->		int:






				"""simple docstring"""



				with self.assertRaises(_snake_case						):
						lowercase__   :    Optional[Any]				=			pa.array(TypedSequence([1, 2, 3]						)			,type=pa.intaa()						)







		def 					UpperCAmelCase    (	self						:  str						)					->		List[Any]:






				"""simple docstring"""



				with self.assertRaises(_snake_case						):
						lowercase__   :    Dict				=			pa.array(TypedSequence([1, 2, 3]			,try_type=Value('''bool'''						)			,type=Value('''int64'''						)						)						)







		def 					UpperCAmelCase    (	self						:  Any						)					->		Tuple:






				"""simple docstring"""



				lowercase__   :    Union[str, Any]				=			pa.array(TypedSequence([1, 2, 3]			,type=Value('''int32'''						)						)						)
				self.assertEqual(arr.type			,pa.intaa()						)







		def 					UpperCAmelCase    (	self						:  Optional[int]						)					->		Optional[int]:






				"""simple docstring"""



				with self.assertRaises((TypeError, pa.lib.ArrowInvalid)						):
						lowercase__   :    List[str]				=			pa.array(TypedSequence(['''foo''', '''bar''']			,type=Value('''int64'''						)						)						)







		def 					UpperCAmelCase    (	self						:  Optional[int]						)					->		Any:






				"""simple docstring"""



				lowercase__   :    Union[str, Any]				=			pa.array(TypedSequence([1, 2, 3]			,try_type=Value('''int32'''						)						)						)
				self.assertEqual(arr.type			,pa.intaa()						)







		def 					UpperCAmelCase    (	self						:  Optional[Any]						)					->		Optional[Any]:






				"""simple docstring"""



				lowercase__   :    Dict				=			pa.array(TypedSequence(['''foo''', '''bar''']			,try_type=Value('''int64'''						)						)						)
				self.assertEqual(arr.type			,pa.string()						)







		def 					UpperCAmelCase    (	self						:  Tuple						)					->		List[str]:






				"""simple docstring"""



				lowercase__   :    str				=			pa.array(TypedSequence([[[1, 2, 3]]]			,type=ArrayaD((1, 3)			,'''int64'''						)						)						)
				self.assertEqual(arr.type			,ArrayaDExtensionType((1, 3)			,'''int64'''						)						)







		def 					UpperCAmelCase    (	self						:  Optional[int]						)					->		str:






				"""simple docstring"""



				with self.assertRaises((TypeError, pa.lib.ArrowInvalid)						):
						lowercase__   :    List[str]				=			pa.array(TypedSequence(['''foo''', '''bar''']			,type=ArrayaD((1, 3)			,'''int64'''						)						)						)







		def 					UpperCAmelCase    (	self						:  Optional[Any]						)					->		Optional[Any]:






				"""simple docstring"""



				lowercase__   :    Tuple				=			pa.array(TypedSequence([[[1, 2, 3]]]			,try_type=ArrayaD((1, 3)			,'''int64'''						)						)						)
				self.assertEqual(arr.type			,ArrayaDExtensionType((1, 3)			,'''int64'''						)						)







		def 					UpperCAmelCase    (	self						:  Tuple						)					->		str:






				"""simple docstring"""



				lowercase__   :    List[Any]				=			pa.array(TypedSequence(['''foo''', '''bar''']			,try_type=ArrayaD((1, 3)			,'''int64'''						)						)						)
				self.assertEqual(arr.type			,pa.string()						)







		@require_pil
		def 					UpperCAmelCase    (	self						:  Any						)					->		int:






				"""simple docstring"""



				import PIL.Image

				lowercase__   :    List[Any]				=			PIL.Image.fromarray(np.arange(10			,dtype=np.uinta						).reshape(2			,5						)						)
				with patch(
				    '''datasets.arrow_writer.cast_to_python_objects'''			,side_effect=_snake_case						) as mock_cast_to_python_objects:
						lowercase__   :    int				=			pa.array(TypedSequence([{'''path''': None, '''bytes''': b'''image_bytes'''}, pil_image]			,type=Image()						)						)
						lowercase__		,			lowercase__   :    Tuple				=			mock_cast_to_python_objects.call_args_list[-1]
						self.assertIn('''optimize_list_casting'''			,_snake_case						)
						self.assertFalse(kwargs['''optimize_list_casting''']						)





def      __UpperCAmelCase      (    __lowerCamelCase						,     __lowerCamelCase	)							->      str:
		lowercase__   :    Optional[int]				=			pa.BufferReader(__lowerCamelCase	) if isinstance(__lowerCamelCase						,     pa.Buffer	) else pa.memory_map(__lowerCamelCase	)
		lowercase__   :    Dict				=			pa.ipc.open_stream(__lowerCamelCase	)
		lowercase__   :    pa.Table				=			f.read_all()
		assert len(pa_table.to_batches()	) == expected_num_chunks
		assert pa_table.to_pydict() == {"col_1": ["foo", "bar"], "col_2": [1, 2]}
		del pa_table



@pytest.mark.parametrize('''writer_batch_size'''						,     [None, 1, 10]	)
@pytest.mark.parametrize(
    '''fields'''						,     [None, {'''col_1''': pa.string(), '''col_2''': pa.intaa()}, {'''col_1''': pa.string(), '''col_2''': pa.intaa()}]	)
def      __UpperCAmelCase      (    __lowerCamelCase						,     __lowerCamelCase	)							->      List[Any]:
		lowercase__   :    List[Any]				=			pa.BufferOutputStream()
		lowercase__   :    Dict				=			pa.schema(__lowerCamelCase	) if fields else None
		with ArrowWriter(stream=__lowerCamelCase						,     schema=__lowerCamelCase						,     writer_batch_size=__lowerCamelCase	) as writer:
				writer.write({'''col_1''': '''foo''', '''col_2''': 1}	)
				writer.write({'''col_1''': '''bar''', '''col_2''': 2}	)
				lowercase__		,			lowercase__   :    Tuple				=			writer.finalize()
		assert num_examples == 2
		assert num_bytes > 0
		if not fields:
				lowercase__   :    Any				=			{'''col_1''': pa.string(), '''col_2''': pa.intaa()}
		assert writer._schema == pa.schema(__lowerCamelCase						,     metadata=writer._schema.metadata	)
		_check_output(output.getvalue()						,     expected_num_chunks=num_examples if writer_batch_size == 1 else 1	)



def      __UpperCAmelCase      (    )							->      List[str]:
		lowercase__   :    List[Any]				=			pa.BufferOutputStream()
		lowercase__   :    str				=			Features({'''labels''': ClassLabel(names=['''neg''', '''pos''']	)}	)
		with ArrowWriter(stream=__lowerCamelCase						,     features=__lowerCamelCase	) as writer:
				writer.write({'''labels''': 0}	)
				writer.write({'''labels''': 1}	)
				lowercase__		,			lowercase__   :    Optional[Any]				=			writer.finalize()
		assert num_examples == 2
		assert num_bytes > 0
		assert writer._schema == features.arrow_schema
		assert writer._schema.metadata == features.arrow_schema.metadata
		lowercase__   :    Optional[int]				=			pa.BufferReader(output.getvalue()	)
		lowercase__   :    Tuple				=			pa.ipc.open_stream(__lowerCamelCase	)
		lowercase__   :    pa.Table				=			f.read_all()
		lowercase__   :    Dict				=			pa_table.schema
		assert pa_table.num_rows == 2
		assert schema == features.arrow_schema
		assert schema.metadata == features.arrow_schema.metadata
		assert features == Features.from_arrow_schema(__lowerCamelCase	)



@pytest.mark.parametrize('''writer_batch_size'''						,     [None, 1, 10]	)
def      __UpperCAmelCase      (    __lowerCamelCase	)							->      List[Any]:
		lowercase__   :    Optional[int]				=			pa.BufferOutputStream()
		with ArrowWriter(
		    stream=__lowerCamelCase						,     writer_batch_size=__lowerCamelCase						,     hash_salt='''split_name'''						,     check_duplicates=__lowerCamelCase						,     ) as writer:
				with pytest.raises(__lowerCamelCase	):
						writer.write({'''col_1''': '''foo''', '''col_2''': 1}						,     key=[1, 2]	)
						lowercase__		,			lowercase__   :    Dict				=			writer.finalize()



@pytest.mark.parametrize('''writer_batch_size'''						,     [None, 2, 10]	)
def      __UpperCAmelCase      (    __lowerCamelCase	)							->      Any:
		lowercase__   :    str				=			pa.BufferOutputStream()
		with ArrowWriter(
		    stream=__lowerCamelCase						,     writer_batch_size=__lowerCamelCase						,     hash_salt='''split_name'''						,     check_duplicates=__lowerCamelCase						,     ) as writer:
				with pytest.raises(__lowerCamelCase	):
						writer.write({'''col_1''': '''foo''', '''col_2''': 1}						,     key=10	)
						writer.write({'''col_1''': '''bar''', '''col_2''': 2}						,     key=10	)
						lowercase__		,			lowercase__   :    str				=			writer.finalize()



@pytest.mark.parametrize('''writer_batch_size'''						,     [None, 2, 10]	)
def      __UpperCAmelCase      (    __lowerCamelCase	)							->      Optional[Any]:
		lowercase__   :    str				=			pa.BufferOutputStream()
		with ArrowWriter(
		    stream=__lowerCamelCase						,     writer_batch_size=__lowerCamelCase						,     hash_salt='''split_name'''						,     check_duplicates=__lowerCamelCase						,     ) as writer:
				writer.write({'''col_1''': '''foo''', '''col_2''': 1}						,     key=1	)
				writer.write({'''col_1''': '''bar''', '''col_2''': 2}						,     key=2	)
				lowercase__		,			lowercase__   :    List[str]				=			writer.finalize()
		assert num_examples == 2
		assert num_bytes > 0
		_check_output(output.getvalue()						,     expected_num_chunks=num_examples if writer_batch_size == 1 else 1	)



@pytest.mark.parametrize('''writer_batch_size'''						,     [None, 1, 10]	)
@pytest.mark.parametrize(
    '''fields'''						,     [None, {'''col_1''': pa.string(), '''col_2''': pa.intaa()}, {'''col_1''': pa.string(), '''col_2''': pa.intaa()}]	)
def      __UpperCAmelCase      (    __lowerCamelCase						,     __lowerCamelCase	)							->      List[Any]:
		lowercase__   :    int				=			pa.BufferOutputStream()
		lowercase__   :    Tuple				=			pa.schema(__lowerCamelCase	) if fields else None
		with ArrowWriter(stream=__lowerCamelCase						,     schema=__lowerCamelCase						,     writer_batch_size=__lowerCamelCase	) as writer:
				writer.write_batch({'''col_1''': ['''foo''', '''bar'''], '''col_2''': [1, 2]}	)
				writer.write_batch({'''col_1''': [], '''col_2''': []}	)
				lowercase__		,			lowercase__   :    List[Any]				=			writer.finalize()
		assert num_examples == 2
		assert num_bytes > 0
		if not fields:
				lowercase__   :    Tuple				=			{'''col_1''': pa.string(), '''col_2''': pa.intaa()}
		assert writer._schema == pa.schema(__lowerCamelCase						,     metadata=writer._schema.metadata	)
		_check_output(output.getvalue()						,     expected_num_chunks=num_examples if writer_batch_size == 1 else 1	)



@pytest.mark.parametrize('''writer_batch_size'''						,     [None, 1, 10]	)
@pytest.mark.parametrize(
    '''fields'''						,     [None, {'''col_1''': pa.string(), '''col_2''': pa.intaa()}, {'''col_1''': pa.string(), '''col_2''': pa.intaa()}]	)
def      __UpperCAmelCase      (    __lowerCamelCase						,     __lowerCamelCase	)							->      Optional[int]:
		lowercase__   :    Dict				=			pa.BufferOutputStream()
		lowercase__   :    Tuple				=			pa.schema(__lowerCamelCase	) if fields else None
		with ArrowWriter(stream=__lowerCamelCase						,     schema=__lowerCamelCase						,     writer_batch_size=__lowerCamelCase	) as writer:
				writer.write_table(pa.Table.from_pydict({'''col_1''': ['''foo''', '''bar'''], '''col_2''': [1, 2]}	)	)
				lowercase__		,			lowercase__   :    Any				=			writer.finalize()
		assert num_examples == 2
		assert num_bytes > 0
		if not fields:
				lowercase__   :    Any				=			{'''col_1''': pa.string(), '''col_2''': pa.intaa()}
		assert writer._schema == pa.schema(__lowerCamelCase						,     metadata=writer._schema.metadata	)
		_check_output(output.getvalue()						,     expected_num_chunks=num_examples if writer_batch_size == 1 else 1	)



@pytest.mark.parametrize('''writer_batch_size'''						,     [None, 1, 10]	)
@pytest.mark.parametrize(
    '''fields'''						,     [None, {'''col_1''': pa.string(), '''col_2''': pa.intaa()}, {'''col_1''': pa.string(), '''col_2''': pa.intaa()}]	)
def      __UpperCAmelCase      (    __lowerCamelCase						,     __lowerCamelCase	)							->      Optional[Any]:
		lowercase__   :    Dict				=			pa.BufferOutputStream()
		lowercase__   :    Dict				=			pa.schema(__lowerCamelCase	) if fields else None
		with ArrowWriter(stream=__lowerCamelCase						,     schema=__lowerCamelCase						,     writer_batch_size=__lowerCamelCase	) as writer:
				writer.write_row(pa.Table.from_pydict({'''col_1''': ['''foo'''], '''col_2''': [1]}	)	)
				writer.write_row(pa.Table.from_pydict({'''col_1''': ['''bar'''], '''col_2''': [2]}	)	)
				lowercase__		,			lowercase__   :    Dict				=			writer.finalize()
		assert num_examples == 2
		assert num_bytes > 0
		if not fields:
				lowercase__   :    Optional[int]				=			{'''col_1''': pa.string(), '''col_2''': pa.intaa()}
		assert writer._schema == pa.schema(__lowerCamelCase						,     metadata=writer._schema.metadata	)
		_check_output(output.getvalue()						,     expected_num_chunks=num_examples if writer_batch_size == 1 else 1	)



def      __UpperCAmelCase      (    )							->      int:
		with tempfile.TemporaryDirectory() as tmp_dir:
				lowercase__   :    Optional[int]				=			{'''col_1''': pa.string(), '''col_2''': pa.intaa()}
				lowercase__   :    Dict				=			os.path.join(__lowerCamelCase						,     '''test.arrow'''	)
				with ArrowWriter(path=__lowerCamelCase						,     schema=pa.schema(__lowerCamelCase	)	) as writer:
						writer.write_batch({'''col_1''': ['''foo''', '''bar'''], '''col_2''': [1, 2]}	)
						lowercase__		,			lowercase__   :    str				=			writer.finalize()
				assert num_examples == 2
				assert num_bytes > 0
				assert writer._schema == pa.schema(__lowerCamelCase						,     metadata=writer._schema.metadata	)
				_check_output(__lowerCamelCase						,     1	)



def      __UpperCAmelCase      (    __lowerCamelCase	)							->      Tuple:
		if pa.types.is_list(__lowerCamelCase	):
				return get_base_dtype(arr_type.value_type	)
		else:
				return arr_type



def      __UpperCAmelCase      (    __lowerCamelCase						,     __lowerCamelCase	)							->      Dict:
		if isinstance(lst[0]						,     __lowerCamelCase	):
				change_first_primitive_element_in_list(lst[0]						,     __lowerCamelCase	)
		else:
				lowercase__   :    Optional[Any]				=			value



@pytest.mark.parametrize('''optimized_int_type, expected_dtype'''						,     [(None, pa.intaa()), (Value('''int32'''	), pa.intaa())]	)
@pytest.mark.parametrize('''sequence'''						,     [[1, 2, 3], [[1, 2, 3]], [[[1, 2, 3]]]]	)
def      __UpperCAmelCase      (    __lowerCamelCase						,     __lowerCamelCase						,     __lowerCamelCase	)							->      Optional[int]:
		lowercase__   :    Optional[int]				=			pa.array(TypedSequence(__lowerCamelCase						,     optimized_int_type=__lowerCamelCase	)	)
		assert get_base_dtype(arr.type	) == expected_dtype



@pytest.mark.parametrize(
    '''col, expected_dtype'''						,     [
        ('''attention_mask''', pa.inta()),
        ('''special_tokens_mask''', pa.inta()),
        ('''token_type_ids''', pa.inta()),
        ('''input_ids''', pa.intaa()),
        ('''other''', pa.intaa()),
    ]						,     )
@pytest.mark.parametrize('''sequence'''						,     [[1, 2, 3], [[1, 2, 3]], [[[1, 2, 3]]]]	)
def      __UpperCAmelCase      (    __lowerCamelCase						,     __lowerCamelCase						,     __lowerCamelCase	)							->      Any:
		# in range
		lowercase__   :    Optional[int]				=			pa.array(OptimizedTypedSequence(__lowerCamelCase						,     col=__lowerCamelCase	)	)
		assert get_base_dtype(arr.type	) == expected_dtype

		# not in range
		if col != "other":
				# avoids errors due to in-place modifications
				lowercase__   :    Tuple				=			copy.deepcopy(__lowerCamelCase	)
				lowercase__   :    Dict				=			np.iinfo(expected_dtype.to_pandas_dtype()	).max + 1
				change_first_primitive_element_in_list(__lowerCamelCase						,     __lowerCamelCase	)
				lowercase__   :    int				=			pa.array(OptimizedTypedSequence(__lowerCamelCase						,     col=__lowerCamelCase	)	)
				assert get_base_dtype(arr.type	) == pa.intaa()



@pytest.mark.parametrize('''raise_exception'''						,     [False, True]	)
def      __UpperCAmelCase      (    __lowerCamelCase						,     __lowerCamelCase	)							->      Optional[int]:
		lowercase__   :    List[Any]				=			str(tmp_path / '''dataset-train.arrow'''	)
		try:
				with ArrowWriter(path=__lowerCamelCase	) as writer:
						if raise_exception:
								raise pa.lib.ArrowInvalid()
						else:
								writer.stream.close()
		except pa.lib.ArrowInvalid:
				pass
		finally:
				assert writer.stream.closed



def      __UpperCAmelCase      (    __lowerCamelCase	)							->      Union[str, Any]:
		lowercase__   :    int				=			'''mock://dataset-train.arrow'''
		with ArrowWriter(path=__lowerCamelCase						,     storage_options=mockfs.storage_options	) as writer:
				assert isinstance(writer._fs						,     type(__lowerCamelCase	)	)
				assert writer._fs.storage_options == mockfs.storage_options
				writer.write({'''col_1''': '''foo''', '''col_2''': 1}	)
				writer.write({'''col_1''': '''bar''', '''col_2''': 2}	)
				lowercase__		,			lowercase__   :    int				=			writer.finalize()
		assert num_examples == 2
		assert num_bytes > 0
		assert mockfs.exists(__lowerCamelCase	)



def      __UpperCAmelCase      (    )							->      Optional[int]:
		lowercase__   :    List[Any]				=			pa.BufferOutputStream()
		with ParquetWriter(stream=__lowerCamelCase	) as writer:
				writer.write({'''col_1''': '''foo''', '''col_2''': 1}	)
				writer.write({'''col_1''': '''bar''', '''col_2''': 2}	)
				lowercase__		,			lowercase__   :    List[str]				=			writer.finalize()
		assert num_examples == 2
		assert num_bytes > 0
		lowercase__   :    str				=			pa.BufferReader(output.getvalue()	)
		lowercase__   :    pa.Table				=			pq.read_table(__lowerCamelCase	)
		assert pa_table.to_pydict() == {"col_1": ["foo", "bar"], "col_2": [1, 2]}



@require_pil
@pytest.mark.parametrize('''embed_local_files'''						,     [False, True]	)
def      __UpperCAmelCase      (    __lowerCamelCase						,     __lowerCamelCase	)							->      Dict:
		import PIL.Image

		lowercase__   :    Dict				=			str(tmp_path / '''test_image_rgb.jpg'''	)
		PIL.Image.fromarray(np.zeros((5, 5)						,     dtype=np.uinta	)	).save(__lowerCamelCase						,     format='''png'''	)
		lowercase__   :    Tuple				=			pa.BufferOutputStream()
		with ParquetWriter(
		    stream=__lowerCamelCase						,     features=Features({'''image''': Image()}	)						,     embed_local_files=__lowerCamelCase	) as writer:
				writer.write({'''image''': image_path}	)
				writer.finalize()
		lowercase__   :    Any				=			pa.BufferReader(output.getvalue()	)
		lowercase__   :    pa.Table				=			pq.read_table(__lowerCamelCase	)
		lowercase__   :    Union[str, Any]				=			pa_table.to_pydict()
		if embed_local_files:
				assert isinstance(out['''image'''][0]['''path''']						,     __lowerCamelCase	)
				with open(__lowerCamelCase						,     '''rb'''	) as f:
						assert out["image"][0]["bytes"] == f.read()
		else:
				assert out["image"][0]["path"] == image_path
				assert out["image"][0]["bytes"] is None





def      __UpperCAmelCase      (    )							->      Union[str, Any]:
		lowercase__   :    Optional[Any]				=			pa.schema([pa.field('''col_1'''						,     pa.string()						,     nullable=__lowerCamelCase	)]	)
		lowercase__   :    List[Any]				=			pa.BufferOutputStream()
		with ArrowWriter(stream=__lowerCamelCase	) as writer:
				writer._build_writer(inferred_schema=__lowerCamelCase	)
				assert writer._schema == pa.schema([pa.field('''col_1'''						,     pa.string()	)]	)




'''simple docstring'''






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


__A					:  Optional[Any]											=      datasets.utils.logging.get_logger(__name__)

if TYPE_CHECKING:
  import pyspark






@dataclass
class    __snake_case       (   datasets.BuilderConfig):







    """simple docstring"""



    lowercase    =					None







def 				UpperCamelCase_			(    A__					:   "pyspark.sql.DataFrame"							,    A__					:   List[int]							,    ):
  '''simple docstring'''





  import pyspark

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

  return generate_fn






class    __snake_case       (   _BaseExamplesIterable):







    """simple docstring"""



    def __init__(   self    :  Optional[int]    ,						lowerCamelCase    :  "pyspark.sql.DataFrame"    ,						lowerCamelCase    :  List[Any]=None    ,						)   ->  Optional[Any]:
      lowerCAmelCase_			:      str			=   df
      lowerCAmelCase_			:      List[Any]			=   partition_order or range(self.df.rdd.getNumPartitions()		)
      lowerCAmelCase_			:      List[str]			=   _generate_iterable_examples(self.df    ,						self.partition_order		)

    def __iter__(   self    :  Dict		)   ->  int:
      yield from self.generate_examples_fn()

    def __lowercase   (   self    :  Any    ,						lowerCamelCase    :  np.random.Generator		)   ->  "SparkExamplesIterable":
      lowerCAmelCase_			:      Optional[int]			=   list(range(self.df.rdd.getNumPartitions()		)		)
      generator.shuffle(lowerCamelCase		)
      return SparkExamplesIterable(self.df    ,						partition_order=lowerCamelCase		)

    def __lowercase   (   self    :  Optional[Any]    ,						lowerCamelCase    :  int    ,						lowerCamelCase    :  int		)   ->  "SparkExamplesIterable":
      lowerCAmelCase_			:      Tuple			=   self.split_shard_indices_by_worker(lowerCamelCase    ,						lowerCamelCase		)
      return SparkExamplesIterable(self.df    ,						partition_order=lowerCamelCase		)





    @property
    def __lowercase   (   self    :  str		)   ->  int:
      return len(self.partition_order		)







class    __snake_case       (   datasets.DatasetBuilder):







    """simple docstring"""



    lowercase    =					SparkConfig

    def __init__(   self    :  Union[str, Any]    ,						lowerCamelCase    :  "pyspark.sql.DataFrame"    ,						lowerCamelCase    :  str = None    ,						lowerCamelCase    :  str = None    ,						**lowerCamelCase    :  int    ,						)   ->  Tuple:
      import pyspark

      lowerCAmelCase_			:      Dict			=   pyspark.sql.SparkSession.builder.getOrCreate()
      lowerCAmelCase_			:      List[str]			=   df
      lowerCAmelCase_			:      List[Any]			=   working_dir

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

    def __lowercase   (   self    :  Tuple		)   ->  Union[str, Any]:
      # Returns the path of the created file.
      def create_cache_and_write_probe(lowerCamelCase    :  int		):
        # 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=lowerCamelCase		)
        lowerCAmelCase_			:      List[str]			=   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(lowerCamelCase    ,						"""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:
        lowerCAmelCase_			:      str			=   (
            self._spark.sparkContext.parallelize(range(1		)    ,						1		).mapPartitions(lowerCamelCase		).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 __lowercase   (   self    :  Any		)   ->  List[Any]:
      return datasets.DatasetInfo(features=self.config.features		)

    def __lowercase   (   self    :  str    ,						lowerCamelCase    :  datasets.download.download_manager.DownloadManager		)   ->  Optional[int]:
      return [datasets.SplitGenerator(name=datasets.Split.TRAIN		)]

    def __lowercase   (   self    :  Any    ,						lowerCamelCase    :  Tuple		)   ->  Optional[Any]:
      import pyspark

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

      lowerCAmelCase_			:      str			=   self.df.count()
      lowerCAmelCase_			:      Any			=   df_num_rows if df_num_rows <= 1_00 else 1_00
      # Approximate the size of each row (in Arrow format) by averaging over a max-100-row sample.
      lowerCAmelCase_			:      str			=   (
          self.df.limit(lowerCamelCase		)
          .repartition(1		)
          .mapInArrow(lowerCamelCase    ,						"""batch_bytes: long"""		)
          .agg(pyspark.sql.functions.sum("""batch_bytes"""		).alias("""sample_bytes"""		)		)
          .collect()[0]
          .sample_bytes
          / sample_num_rows
      )
      lowerCAmelCase_			:      List[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.
        lowerCAmelCase_			:      Tuple			=   min(lowerCamelCase    ,						int(approx_total_size / max_shard_size		)		)
        lowerCAmelCase_			:      Dict			=   self.df.repartition(lowerCamelCase		)

    def __lowercase   (   self    :  Optional[int]    ,						lowerCamelCase    :  str    ,						lowerCamelCase    :  str    ,						lowerCamelCase    :  int    ,						)   ->  Iterable[Tuple[int, bool, Union[int, tuple]]]:
      import pyspark

      lowerCAmelCase_			:      List[str]			=   ParquetWriter if file_format == """parquet""" else ArrowWriter
      lowerCAmelCase_			:      int			=   os.path.join(self._working_dir    ,						os.path.basename(lowerCamelCase		)		) if self._working_dir else fpath
      lowerCAmelCase_			:      Tuple			=   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.
      lowerCAmelCase_			:      str			=   self.config.features
      lowerCAmelCase_			:      Tuple			=   self._writer_batch_size
      lowerCAmelCase_			:      Any			=   self._fs.storage_options

      def write_arrow(lowerCamelCase    :  Optional[int]		):
        # Within the same SparkContext, no two task attempts will share the same attempt ID.
        lowerCAmelCase_			:      Union[str, Any]			=   pyspark.TaskContext().taskAttemptId()
        lowerCAmelCase_			:      Union[str, Any]			=   next(lowerCamelCase    ,						lowerCamelCase		)
        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"""]    ,						)
        lowerCAmelCase_			:      List[Any]			=   0
        lowerCAmelCase_			:      Dict			=   writer_class(
            features=lowerCamelCase    ,						path=working_fpath.replace("""SSSSS"""    ,						F'{shard_id:05d}'		).replace("""TTTTT"""    ,						F'{task_id:05d}'		)    ,						writer_batch_size=lowerCamelCase    ,						storage_options=lowerCamelCase    ,						embed_local_files=lowerCamelCase    ,						)
        lowerCAmelCase_			:      str			=   pa.Table.from_batches([first_batch]		)
        writer.write_table(lowerCamelCase		)
        for batch in it:
          if max_shard_size is not None and writer._num_bytes >= max_shard_size:
            lowerCAmelCase_,							lowerCAmelCase_			:      Any			=   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
            lowerCAmelCase_			:      Optional[int]			=   writer_class(
                features=writer._features    ,						path=working_fpath.replace("""SSSSS"""    ,						F'{shard_id:05d}'		).replace("""TTTTT"""    ,						F'{task_id:05d}'		)    ,						writer_batch_size=lowerCamelCase    ,						storage_options=lowerCamelCase    ,						embed_local_files=lowerCamelCase    ,						)
          lowerCAmelCase_			:      Tuple			=   pa.Table.from_batches([batch]		)
          writer.write_table(lowerCamelCase		)

        if writer._num_bytes > 0:
          lowerCAmelCase_,							lowerCAmelCase_			:      List[str]			=   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(lowerCamelCase		)		):
            lowerCAmelCase_			:      str			=   os.path.join(os.path.dirname(lowerCamelCase		)    ,						os.path.basename(lowerCamelCase		)		)
            shutil.move(lowerCamelCase    ,						lowerCamelCase		)

      lowerCAmelCase_			:      Dict			=   (
          self.df.mapInArrow(lowerCamelCase    ,						"""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 __lowercase   (   self    :  Any    ,						lowerCamelCase    :  "datasets.SplitGenerator"    ,						lowerCamelCase    :  str = "arrow"    ,						lowerCamelCase    :  Optional[Union[str, int]] = None    ,						lowerCamelCase    :  Optional[int] = None    ,						**lowerCamelCase    :  List[str]    ,						)   ->  Optional[int]:
      self._validate_cache_dir()

      lowerCAmelCase_			:      List[str]			=   convert_file_size_to_int(max_shard_size or MAX_SHARD_SIZE		)
      self._repartition_df_if_needed(lowerCamelCase		)
      lowerCAmelCase_			:      Optional[int]			=   not is_remote_filesystem(self._fs		)
      lowerCAmelCase_			:      Dict			=   os.path.join if is_local else posixpath.join

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

      lowerCAmelCase_			:      Dict			=   0
      lowerCAmelCase_			:      int			=   0
      lowerCAmelCase_			:      Dict			=   0
      lowerCAmelCase_			:      Any			=   []
      lowerCAmelCase_			:      Dict			=   []

      for task_id, content in self._prepare_split_single(lowerCamelCase    ,						lowerCamelCase    ,						lowerCamelCase		):
        (
            (
            lowerCAmelCase_
        ),							(
            lowerCAmelCase_
        ),							(
            lowerCAmelCase_
        ),							(
            lowerCAmelCase_
        ),							
        )			:      str			=   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(lowerCamelCase		)

      lowerCAmelCase_			:      Union[str, Any]			=   total_num_examples
      lowerCAmelCase_			:      int			=   total_num_bytes

      # should rename everything at the end
      logger.debug(F'Renaming {total_shards} shards.'		)
      if total_shards > 1:
        lowerCAmelCase_			:      Optional[int]			=   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.
        lowerCAmelCase_			:      List[Any]			=   self._fs

        # use the -SSSSS-of-NNNNN pattern
        def _rename_shard(
            lowerCamelCase    :  int    ,						lowerCamelCase    :  int    ,						lowerCamelCase    :  int    ,						):
          rename(
              lowerCamelCase    ,						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}'		)    ,						)

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





    def __lowercase   (   self    :  Dict    ,						lowerCamelCase    :  "datasets.SplitGenerator"    ,						)   ->  SparkExamplesIterable:
      return SparkExamplesIterable(self.df		)



