
'''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  snake_case_		(    __SCREAMING_SNAKE_CASE :			Union[str, Any]				,				__SCREAMING_SNAKE_CASE :			Optional[int]							):



      """simple docstring"""





      lowercase_     :					Optional[int]													=	[]
      for part_id in partition_order:
            lowercase_     :					int													=	df.where(F'''SPARK_PARTITION_ID() = {part_id}'''							).collect()
            for row_idx, row in enumerate(__SCREAMING_SNAKE_CASE							):
                  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  snake_case_		(    ):



      """simple docstring"""





      lowercase_     :					Optional[int]													=	pyspark.sql.SparkSession.builder.master('''local[*]'''							).appName('''pyspark'''							).getOrCreate()
      lowercase_     :					Optional[int]													=	spark.range(100							).repartition(1							)
      lowercase_     :					List[Any]													=	Spark(__SCREAMING_SNAKE_CASE							)
      # 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  snake_case_		(    ):



      """simple docstring"""





      lowercase_     :					str													=	pyspark.sql.SparkSession.builder.master('''local[*]'''							).appName('''pyspark'''							).getOrCreate()
      lowercase_     :					Optional[Any]													=	spark.range(10							).repartition(2							)
      lowercase_     :					List[Any]													=	[1, 0]
      lowercase_     :					Optional[int]													=	_generate_iterable_examples(__SCREAMING_SNAKE_CASE				,				__SCREAMING_SNAKE_CASE							)  # Reverse the partitions.
      lowercase_     :					List[str]													=	_get_expected_row_ids_and_row_dicts_for_partition_order(__SCREAMING_SNAKE_CASE				,				__SCREAMING_SNAKE_CASE							)

      for i, (row_id, row_dict) in enumerate(generate_fn()							):
            lowercase_     :					Union[str, Any]													=	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  snake_case_		(    ):



      """simple docstring"""





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



      """simple docstring"""





      lowercase_     :					int													=	pyspark.sql.SparkSession.builder.master('''local[*]'''							).appName('''pyspark'''							).getOrCreate()
      lowercase_     :					Any													=	spark.range(30							).repartition(3							)
      # Mock the generator so that shuffle reverses the partition indices.
      with patch('''numpy.random.Generator'''							) as generator_mock:
            lowercase_     :					Optional[int]													=	lambda __SCREAMING_SNAKE_CASE							: x.reverse()
            lowercase_     :					Optional[Any]													=	_get_expected_row_ids_and_row_dicts_for_partition_order(__SCREAMING_SNAKE_CASE				,				[2, 1, 0]							)

            lowercase_     :					Any													=	SparkExamplesIterable(__SCREAMING_SNAKE_CASE							).shuffle_data_sources(__SCREAMING_SNAKE_CASE							)
            assert shuffled_it.n_shards == 3
            for i, (row_id, row_dict) in enumerate(__SCREAMING_SNAKE_CASE							):
                  lowercase_     :					Tuple													=	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  snake_case_		(    ):



      """simple docstring"""





      lowercase_     :					Tuple													=	pyspark.sql.SparkSession.builder.master('''local[*]'''							).appName('''pyspark'''							).getOrCreate()
      lowercase_     :					Optional[Any]													=	spark.range(20							).repartition(4							)

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

      # Partitions 1 and 3
      lowercase_     :					List[Any]													=	SparkExamplesIterable(__SCREAMING_SNAKE_CASE							).shard_data_sources(worker_id=1				,				num_workers=2							)
      assert shard_it_a.n_shards == 2
      lowercase_     :					int													=	_get_expected_row_ids_and_row_dicts_for_partition_order(__SCREAMING_SNAKE_CASE				,				[1, 3]							)
      for i, (row_id, row_dict) in enumerate(__SCREAMING_SNAKE_CASE							):
            lowercase_     :					Optional[int]													=	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  snake_case_		(    ):



      """simple docstring"""





      lowercase_     :					List[str]													=	pyspark.sql.SparkSession.builder.master('''local[*]'''							).appName('''pyspark'''							).getOrCreate()
      lowercase_     :					int													=	spark.range(100							).repartition(1							)
      lowercase_     :					Union[str, Any]													=	Spark(__SCREAMING_SNAKE_CASE							)
      # 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() == 100





'''simple docstring'''







import logging
import os

from .state import PartialState






class 			lowerCAmelCase__       (  logging.LoggerAdapter					):






					@staticmethod
					def      _snake_case   (       __SCREAMING_SNAKE_CASE		):


											"""simple docstring"""





											lowercase_     :					Optional[Any]													=	PartialState()
											return not main_process_only or (main_process_only and state.is_main_process)






					def      _snake_case   (       self    ,							__SCREAMING_SNAKE_CASE    ,							__SCREAMING_SNAKE_CASE    ,							*__SCREAMING_SNAKE_CASE    ,							**__SCREAMING_SNAKE_CASE		):


											"""simple docstring"""





											if PartialState._shared_state == {}:
																	raise RuntimeError(
																	    '''You must initialize the accelerate state by calling either `PartialState()` or `Accelerator()` before using the logging utility.'''		)
											lowercase_     :					Tuple													=	kwargs.pop('''main_process_only'''    ,							__SCREAMING_SNAKE_CASE		)
											lowercase_     :					Optional[int]													=	kwargs.pop('''in_order'''    ,							__SCREAMING_SNAKE_CASE		)

											if self.isEnabledFor(__SCREAMING_SNAKE_CASE		):
																	if self._should_log(__SCREAMING_SNAKE_CASE		):
																							lowercase_						,			lowercase_     :					Optional[Any]													=	self.process(__SCREAMING_SNAKE_CASE    ,							__SCREAMING_SNAKE_CASE		)
																							self.logger.log(__SCREAMING_SNAKE_CASE    ,							__SCREAMING_SNAKE_CASE    ,							*__SCREAMING_SNAKE_CASE    ,							**__SCREAMING_SNAKE_CASE		)

																	elif in_order:
																							lowercase_     :					Optional[Any]													=	PartialState()
																							for i in range(state.num_processes		):
																													if i == state.process_index:
																																			lowercase_						,			lowercase_     :					Optional[int]													=	self.process(__SCREAMING_SNAKE_CASE    ,							__SCREAMING_SNAKE_CASE		)
																																			self.logger.log(__SCREAMING_SNAKE_CASE    ,							__SCREAMING_SNAKE_CASE    ,							*__SCREAMING_SNAKE_CASE    ,							**__SCREAMING_SNAKE_CASE		)
																													state.wait_for_everyone()







def  snake_case_		(    __SCREAMING_SNAKE_CASE :			str				,				__SCREAMING_SNAKE_CASE :			str = None							):



						"""simple docstring"""





						if log_level is None:
												lowercase_     :					Any													=	os.environ.get('''ACCELERATE_LOG_LEVEL'''				,				__SCREAMING_SNAKE_CASE							)
						lowercase_     :					List[str]													=	logging.getLogger(__SCREAMING_SNAKE_CASE							)
						if log_level is not None:
												logger.setLevel(log_level.upper()							)
												logger.root.setLevel(log_level.upper()							)
						return MultiProcessAdapter(__SCREAMING_SNAKE_CASE				,				{}							)



