


'''simple docstring'''



import multiprocessing
import os
from typing import BinaryIO, Optional, Union

import fsspec

from .. import Dataset, Features, NamedSplit, config
from ..formatting import query_table
from ..packaged_modules.json.json import Json
from ..utils import logging
from ..utils.typing import NestedDataStructureLike, PathLike
from .abc import AbstractDatasetReader





class 				__UpperCamelCase							(						UpperCamelCase__   ):




	def __init__(  self		,							__a		,							__a = None		,							__a = None		,							__a = None		,							__a = False		,							__a = False		,							__a = None		,							__a = None		,							**__a		,							):

						'''simple docstring'''





						super().__init__(
						    __a		,							split=__a		,							features=__a		,							cache_dir=__a		,							keep_in_memory=__a		,							streaming=__a		,							num_proc=__a		,							**__a		,							)
						__a     :			int          =    field
						__a     :			Union[str, Any]          =    path_or_paths if isinstance(__a		,							__a				) else {self.split: path_or_paths}
						__a     :			Union[str, Any]          =    Json(
						    cache_dir=__a		,							data_files=__a		,							features=__a		,							field=__a		,							**__a		,							)





	def  __UpperCAmelCase				(  self				):

						'''simple docstring'''





						if self.streaming:
											__a     :			Any          =    self.builder.as_streaming_dataset(split=self.split				)
						# Build regular (map-style) dataset
						else:
											__a     :			Any          =    None
											__a     :			int          =    None
											__a     :			Tuple          =    None
											__a     :			List[Any]          =    None

											self.builder.download_and_prepare(
											    download_config=__a		,							download_mode=__a		,							verification_mode=__a		,							base_path=__a		,							num_proc=self.num_proc		,							)
											__a     :			str          =    self.builder.as_dataset(
											    split=self.split		,							verification_mode=__a		,							in_memory=self.keep_in_memory				)
						return dataset





class 				__UpperCamelCase							:




	def __init__(  self		,							__a		,							__a		,							__a = None		,							__a = None		,							**__a		,							):

						'''simple docstring'''





						if num_proc is not None and num_proc <= 0:
											raise ValueError(f"""num_proc {num_proc} must be an integer > 0."""				)

						__a     :			Optional[int]          =    dataset
						__a     :			List[Any]          =    path_or_buf
						__a     :			Optional[int]          =    batch_size if batch_size else config.DEFAULT_MAX_BATCH_SIZE
						__a     :			Union[str, Any]          =    num_proc
						__a     :			int          =    'utf-8'
						__a     :			str          =    to_json_kwargs




	def  __UpperCAmelCase				(  self				):

						'''simple docstring'''





						__a     :			str          =    self.to_json_kwargs.pop('path_or_buf'		,							__a				)
						__a     :			str          =    self.to_json_kwargs.pop('orient'		,							'records'				)
						__a     :			Union[str, Any]          =    self.to_json_kwargs.pop('lines'		,							True if orient == 'records' else False				)
						__a     :			List[Any]          =    self.to_json_kwargs.pop('index'		,							False if orient in ['split', 'table'] else True				)
						__a     :			Union[str, Any]          =    self.to_json_kwargs.pop('compression'		,							__a				)

						if compression not in [None, "infer", "gzip", "bz2", "xz"]:
											raise NotImplementedError(f"""`datasets` currently does not support {compression} compression"""				)

						if isinstance(self.path_or_buf		,							(str, bytes, os.PathLike)				):
											with fsspec.open(self.path_or_buf		,							'wb'		,							compression=__a				) as buffer:
																__a     :			Tuple          =    self._write(file_obj=__a		,							orient=__a		,							lines=__a		,							index=__a		,							**self.to_json_kwargs				)
						else:
											if compression:
																raise NotImplementedError(
																    f"""The compression parameter is not supported when writing to a buffer, but compression={compression}"""
																    ' was passed. Please provide a local path instead.'				)
											__a     :			Dict          =    self._write(
											    file_obj=self.path_or_buf		,							orient=__a		,							lines=__a		,							index=__a		,							**self.to_json_kwargs				)
						return written




	def  __UpperCAmelCase				(  self		,							__a				):

						'''simple docstring'''





						__a       ,							__a       ,							__a       ,							__a       ,							__a     :			Dict          =    args

						__a     :			int          =    query_table(
						    table=self.dataset.data		,							key=slice(__a		,							offset + self.batch_size				)		,							indices=self.dataset._indices		,							)
						__a     :			List[Any]          =    batch.to_pandas().to_json(
						    path_or_buf=__a		,							orient=__a		,							lines=__a		,							index=__a		,							**__a				)
						if not json_str.endswith('\n'				):
											json_str += "\n"
						return json_str.encode(self.encoding				)





	def  __UpperCAmelCase				(  self		,							__a		,							__a		,							__a		,							__a		,							**__a		,							):

						'''simple docstring'''





						__a     :			Any          =    0

						if self.num_proc is None or self.num_proc == 1:
											for offset in logging.tqdm(
											    range(0		,							len(self.dataset				)		,							self.batch_size				)		,							unit='ba'		,							disable=not logging.is_progress_bar_enabled()		,							desc='Creating json from Arrow format'		,							):
																__a     :			List[Any]          =    self._batch_json((offset, orient, lines, index, to_json_kwargs)				)
																written += file_obj.write(__a				)
						else:
											__a       ,							__a     :			Tuple          =    len(self.dataset				), self.batch_size
											with multiprocessing.Pool(self.num_proc				) as pool:
																for json_str in logging.tqdm(
																    pool.imap(
																        self._batch_json		,							[(offset, orient, lines, index, to_json_kwargs) for offset in range(0		,							__a		,							__a				)]		,							)		,							total=(num_rows // batch_size) + 1 if num_rows % batch_size else num_rows // batch_size		,							unit='ba'		,							disable=not logging.is_progress_bar_enabled()		,							desc='Creating json from Arrow format'		,							):
																					written += file_obj.write(__a				)

						return written





'''simple docstring'''
from unittest.mock import patch

import pyspark

from datasets.packaged_modules.spark.spark import (
    Spark,
    SparkExamplesIterable,
    _generate_iterable_examples,
)

from ..utils import (
    require_dill_gt_0_3_2,
    require_not_windows,
)






def      UpperCamelCase_				(      A__  ,      A__						):
	a_      				=						[]
	for part_id in partition_order:
		a_      				=						df.where(F'''SPARK_PARTITION_ID() = {part_id}'''						).collect()
		for row_idx, row in enumerate(A__						):
			expected_row_ids_and_row_dicts.append((F'''{part_id}_{row_idx}''', row.asDict())						)
	return expected_row_ids_and_row_dicts






@require_not_windows
@require_dill_gt_0_3_2
def      UpperCamelCase_				(      ):
	a_      				=						pyspark.sql.SparkSession.builder.master("""local[*]"""						).appName("""pyspark"""						).getOrCreate()
	a_      				=						spark.range(1_00						).repartition(1						)
	a_      				=						Spark(A__						)
	# The id ints will be converted to Pyarrow int64s, so each row will be 8 bytes. Setting a max_shard_size of 16 means
	# that each partition can hold 2 rows.
	spark_builder._repartition_df_if_needed(max_shard_size=16						)
	# Given that the dataframe has 100 rows and each partition has 2 rows, we expect 50 partitions.
	assert spark_builder.df.rdd.getNumPartitions() == 50






@require_not_windows
@require_dill_gt_0_3_2
def      UpperCamelCase_				(      ):
	a_      				=						pyspark.sql.SparkSession.builder.master("""local[*]"""						).appName("""pyspark"""						).getOrCreate()
	a_      				=						spark.range(10						).repartition(2						)
	a_      				=						[1, 0]
	a_      				=						_generate_iterable_examples(A__  ,      A__						)  # Reverse the partitions.
	a_      				=						_get_expected_row_ids_and_row_dicts_for_partition_order(A__  ,      A__						)

	for i, (row_id, row_dict) in enumerate(generate_fn()						):
		a_				,  a_      				=						expected_row_ids_and_row_dicts[i]
		assert row_id == expected_row_id
		assert row_dict == expected_row_dict






@require_not_windows
@require_dill_gt_0_3_2
def      UpperCamelCase_				(      ):
	a_      				=						pyspark.sql.SparkSession.builder.master("""local[*]"""						).appName("""pyspark"""						).getOrCreate()
	a_      				=						spark.range(10						).repartition(1						)
	a_      				=						SparkExamplesIterable(A__						)
	assert it.n_shards == 1
	for i, (row_id, row_dict) in enumerate(A__						):
		assert row_id == F'''0_{i}'''
		assert row_dict == {"id": i}






@require_not_windows
@require_dill_gt_0_3_2
def      UpperCamelCase_				(      ):
	a_      				=						pyspark.sql.SparkSession.builder.master("""local[*]"""						).appName("""pyspark"""						).getOrCreate()
	a_      				=						spark.range(30						).repartition(3						)
	# Mock the generator so that shuffle reverses the partition indices.
	with patch("""numpy.random.Generator"""						) as generator_mock:
		a_      				=						lambda A__						: x.reverse()
		a_      				=						_get_expected_row_ids_and_row_dicts_for_partition_order(A__  ,      [2, 1, 0]						)

		a_      				=						SparkExamplesIterable(A__						).shuffle_data_sources(A__						)
		assert shuffled_it.n_shards == 3
		for i, (row_id, row_dict) in enumerate(A__						):
			a_				,  a_      				=						expected_row_ids_and_row_dicts[i]
			assert row_id == expected_row_id
			assert row_dict == expected_row_dict






@require_not_windows
@require_dill_gt_0_3_2
def      UpperCamelCase_				(      ):
	a_      				=						pyspark.sql.SparkSession.builder.master("""local[*]"""						).appName("""pyspark"""						).getOrCreate()
	a_      				=						spark.range(20						).repartition(4						)

	# Partitions 0 and 2
	a_      				=						SparkExamplesIterable(A__						).shard_data_sources(worker_id=0  ,      num_workers=2						)
	assert shard_it_a.n_shards == 2
	a_      				=						_get_expected_row_ids_and_row_dicts_for_partition_order(A__  ,      [0, 2]						)
	for i, (row_id, row_dict) in enumerate(A__						):
		a_				,  a_      				=						expected_row_ids_and_row_dicts_a[i]
		assert row_id == expected_row_id
		assert row_dict == expected_row_dict

	# Partitions 1 and 3
	a_      				=						SparkExamplesIterable(A__						).shard_data_sources(worker_id=1  ,      num_workers=2						)
	assert shard_it_a.n_shards == 2
	a_      				=						_get_expected_row_ids_and_row_dicts_for_partition_order(A__  ,      [1, 3]						)
	for i, (row_id, row_dict) in enumerate(A__						):
		a_				,  a_      				=						expected_row_ids_and_row_dicts_a[i]
		assert row_id == expected_row_id
		assert row_dict == expected_row_dict






@require_not_windows
@require_dill_gt_0_3_2
def      UpperCamelCase_				(      ):
	a_      				=						pyspark.sql.SparkSession.builder.master("""local[*]"""						).appName("""pyspark"""						).getOrCreate()
	a_      				=						spark.range(1_00						).repartition(1						)
	a_      				=						Spark(A__						)
	# Choose a small max_shard_size for maximum partitioning.
	spark_builder._repartition_df_if_needed(max_shard_size=1						)
	# The new number of partitions should not be greater than the number of rows.
	assert spark_builder.df.rdd.getNumPartitions() == 1_00
