





'''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


snake_case_     =	datasets.utils.logging.get_logger(__name__)

if TYPE_CHECKING:
     import pyspark


@dataclass
class   _lowercase						(							datasets.BuilderConfig				):
   _UpperCamelCase       		=   None







def 							_lowerCamelCase(  UpperCamelCase__    : "pyspark.sql.DataFrame"    ,       UpperCamelCase__    : List[int]    ,       )		->  Any:
    import pyspark

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

    return generate_fn


class   _lowercase						(							_BaseExamplesIterable				):


   def __init__( self    ,   _UpperCAmelCase    ,   _UpperCAmelCase=None    ,   ):
       A       :       Tuple							=						df
       A       :       str							=						partition_order or range(self.df.rdd.getNumPartitions()					)
       A       :       Any							=						_generate_iterable_examples(self.df    ,   self.partition_order					)


   def __iter__( self					):
       yield from self.generate_examples_fn()


   def 					snake_case				( self    ,   _UpperCAmelCase					):
       A       :       Optional[int]							=						list(range(self.df.rdd.getNumPartitions()					)					)
       generator.shuffle(_UpperCAmelCase					)
       return SparkExamplesIterable(self.df    ,   partition_order=_UpperCAmelCase					)


   def 					snake_case				( self    ,   _UpperCAmelCase    ,   _UpperCAmelCase					):
       A       :       str							=						self.split_shard_indices_by_worker(_UpperCAmelCase    ,   _UpperCAmelCase					)
       return SparkExamplesIterable(self.df    ,   partition_order=_UpperCAmelCase					)


   @property
   def 					snake_case				( self					):
       return len(self.partition_order					)







class   _lowercase						(							datasets.DatasetBuilder				):
   _UpperCamelCase       		=   SparkConfig


   def __init__( self    ,   _UpperCAmelCase    ,   _UpperCAmelCase = None    ,   _UpperCAmelCase = None    ,   **_UpperCAmelCase    ,   ):
       import pyspark

       A       :       Union[str, Any]							=						pyspark.sql.SparkSession.builder.getOrCreate()
       A       :       Any							=						df
       A       :       List[Any]							=						working_dir

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


   def 					snake_case				( self					):
       # Returns the path of the created file.
       def create_cache_and_write_probe(_UpperCAmelCase					):
           # 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					)
           A       :       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:
           A       :       Dict							=						(
               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 					snake_case				( self					):
       return datasets.DatasetInfo(features=self.config.features					)


   def 					snake_case				( self    ,   _UpperCAmelCase					):
       return [datasets.SplitGenerator(name=datasets.Split.TRAIN					)]


   def 					snake_case				( self    ,   _UpperCAmelCase					):
       import pyspark

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

       A       :       List[str]							=						self.df.count()
       A       :       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.
       A       :       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
       )
       A       :       List[str]							=						approx_bytes_per_row * df_num_rows
       if approx_total_size > max_shard_size:
           # Make sure there is at least one row per partition.
           A       :       Any							=						min(_UpperCAmelCase    ,   int(approx_total_size / max_shard_size					)					)
           A       :       int							=						self.df.repartition(_UpperCAmelCase					)


   def 					snake_case				( self    ,   _UpperCAmelCase    ,   _UpperCAmelCase    ,   _UpperCAmelCase    ,   ):
       import pyspark

       A       :       List[Any]							=						ParquetWriter if file_format == '''parquet''' else ArrowWriter
       A       :       List[str]							=						os.path.join(self._working_dir    ,   os.path.basename(_UpperCAmelCase					)					) if self._working_dir else fpath
       A       :       Optional[Any]							=						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.
       A       :       Any							=						self.config.features
       A       :       List[Any]							=						self._writer_batch_size
       A       :       Dict							=						self._fs.storage_options

       def write_arrow(_UpperCAmelCase					):
           # Within the same SparkContext, no two task attempts will share the same attempt ID.
           A       :       Optional[int]							=						pyspark.TaskContext().taskAttemptId()
           A       :       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''']    ,   )
           A       :       Tuple							=						0
           A       :       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    ,   )
           A       :       int							=						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:
                   A       :       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
                   A       :       int							=						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    ,   )
               A       :       Any							=						pa.Table.from_batches([batch]					)
               writer.write_table(_UpperCAmelCase					)

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

       A       :       List[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 					snake_case				( self    ,   _UpperCAmelCase    ,   _UpperCAmelCase = "arrow"    ,   _UpperCAmelCase = None    ,   _UpperCAmelCase = None    ,   **_UpperCAmelCase    ,   ):
       self._validate_cache_dir()

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

       A       :       Any							=						'''-TTTTT-SSSSS-of-NNNNN'''
       A       :       Any							=						f'''{self.name}-{split_generator.name}{SUFFIX}.{file_format}'''
       A       :       List[Any]							=						path_join(self._output_dir    ,   _UpperCAmelCase					)

       A       :       Union[str, Any]							=						0
       A       :       Dict							=						0
       A       :       Union[str, Any]							=						0
       A       :       Optional[Any]							=						[]
       A       :       Optional[Any]							=						[]

       for task_id, content in self._prepare_split_single(_UpperCAmelCase    ,   _UpperCAmelCase    ,   _UpperCAmelCase					):
           (
               A	
           )       :       int							=						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					)

       A       :       List[str]							=						total_num_examples
       A       :       List[str]							=						total_num_bytes

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

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

           A       :       List[Any]							=						[]
           A       :       int							=						0
           for i in range(len(_UpperCAmelCase					)					):
               A       :       Optional[Any]							=						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 _UpperCAmelCase					: _rename_shard(*_UpperCAmelCase					)					).collect()
       else:
           # don't use any pattern
           A       :       Optional[Any]							=						0
           A       :       List[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 					snake_case				( self    ,   _UpperCAmelCase    ,   ):
       return SparkExamplesIterable(self.df					)








'''simple docstring'''






def 							_lowerCamelCase(  UpperCamelCase__    : int		)		->  None:
    A       :       List[Any]							=						generate_pascal_triangle(UpperCamelCase__		)
    for row_idx in range(UpperCamelCase__		):
        # Print left spaces
        for _ in range(num_rows - row_idx - 1		):
            print(end=''' '''		)
        # Print row values
        for col_idx in range(row_idx + 1		):
            if col_idx != row_idx:
                print(triangle[row_idx][col_idx]    ,       end=''' '''		)
            else:
                print(triangle[row_idx][col_idx]    ,       end=''''''		)
        print()
def 							_lowerCamelCase(  UpperCamelCase__    : int		)		->  list[list[int]]:

    if not isinstance(UpperCamelCase__    ,       UpperCamelCase__		):
        raise TypeError('''The input value of \'num_rows\' should be \'int\''''		)

    if num_rows == 0:
        return []
    elif num_rows < 0:
        raise ValueError(
            '''The input value of \'num_rows\' should be greater than or equal to 0'''		)

    A       :       list[list[int]]							=						[]
    for current_row_idx in range(UpperCamelCase__		):
        A       :       Optional[int]							=						populate_current_row(UpperCamelCase__    ,       UpperCamelCase__		)
        triangle.append(UpperCamelCase__		)
    return triangle
def 							_lowerCamelCase(  UpperCamelCase__    : list[list[int]]    ,       UpperCamelCase__    : int		)		->  list[int]:
    A       :       Dict							=						[-1] * (current_row_idx + 1)
    # first and last elements of current row are equal to 1
    A,	A       :       Union[str, Any]							=						1, 1
    for current_col_idx in range(1    ,       UpperCamelCase__		):
        calculate_current_element(
            UpperCamelCase__    ,       UpperCamelCase__    ,       UpperCamelCase__    ,       UpperCamelCase__		)
    return current_row
def 							_lowerCamelCase(  UpperCamelCase__    : list[list[int]]    ,       UpperCamelCase__    : list[int]    ,       UpperCamelCase__    : int    ,       UpperCamelCase__    : int    ,       )		->  None:
    A       :       Dict							=						triangle[current_row_idx - 1][current_col_idx - 1]
    A       :       Any							=						triangle[current_row_idx - 1][current_col_idx]
    A       :       Optional[int]							=						above_to_left_elt + above_to_right_elt
def 							_lowerCamelCase(  UpperCamelCase__    : int		)		->  list[list[int]]:

    if not isinstance(UpperCamelCase__    ,       UpperCamelCase__		):
        raise TypeError('''The input value of \'num_rows\' should be \'int\''''		)

    if num_rows == 0:
        return []
    elif num_rows < 0:
        raise ValueError(
            '''The input value of \'num_rows\' should be greater than or equal to 0'''		)

    A       :       list[list[int]]							=						[[1]]

    for row_index in range(1    ,       UpperCamelCase__		):
        A       :       str							=						[0] + result[-1] + [0]
        A       :       Optional[Any]							=						row_index + 1
        # Calculate the number of distinct elements in a row
        A       :       str							=						sum(divmod(UpperCamelCase__    ,       2		)		)
        A       :       Optional[int]							=						[
            temp_row[i - 1] + temp_row[i] for i in range(1    ,       distinct_elements + 1		)
        ]
        A       :       Any							=						row_first_half[: (row_index + 1) // 2]
        row_second_half.reverse()
        A       :       Any							=						row_first_half + row_second_half
        result.append(UpperCamelCase__		)

    return result
def 							_lowerCamelCase(  )		->  None:
    from collections.abc import Callable
    from timeit import timeit

    def benchmark_a_function(UpperCamelCase__    : Callable    ,       UpperCamelCase__    : int		) -> None:
        A       :       Dict							=						F'''{func.__name__}({value})'''
        A       :       List[Any]							=						timeit(F'''__main__.{call}'''    ,       setup='''import __main__'''		)
        # print(f"{call:38} = {func(value)} -- {timing:.4f} seconds")
        print(F'''{call:38} -- {timing:.4f} seconds'''		)

    for value in range(15		):  # (1, 7, 14):
        for func in (generate_pascal_triangle, generate_pascal_triangle_optimized):
            benchmark_a_function(UpperCamelCase__    ,       UpperCamelCase__		)
        print()


if __name__ == "__main__":
     import doctest

     doctest.testmod()
     benchmark()
