


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


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

if TYPE_CHECKING:
						import pyspark






@dataclass
class 		lowerCAmelCase__	(	datasets.BuilderConfig		):
			lowerCAmelCase       :		Optional[datasets.Features]				      =   None






def        __lowerCAmelCase       (__lowerCAmelCase					,       __lowerCAmelCase					,       ):
		import pyspark

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

		return generate_fn






class 		lowerCAmelCase__	(	_BaseExamplesIterable		):




			def __init__(						self     :   List[Any]		,     lowerCamelCase__     :   List[str]		,     lowerCamelCase__     :   str=None		,     )      ->List[str]:



					'''simple docstring'''







					_UpperCAmelCase						:				str	    =					df
					_UpperCAmelCase						:				Union[str, Any]	    =					partition_order or range(self.df.rdd.getNumPartitions()       )
					_UpperCAmelCase						:				List[Any]	    =					_generate_iterable_examples(self.df		,     self.partition_order       )




			def __iter__(						self     :   List[Any]       )      ->Any:



					'''simple docstring'''







					yield from self.generate_examples_fn()




			def 							lowerCAmelCase__			(						self     :   Tuple		,     lowerCamelCase__     :   str       )      ->Any:



					'''simple docstring'''







					_UpperCAmelCase						:				Optional[Any]	    =					list(range(self.df.rdd.getNumPartitions()       )       )
					generator.shuffle(lowerCamelCase__       )
					return SparkExamplesIterable(self.df		,     partition_order=lowerCamelCase__       )




			def 							lowerCAmelCase__			(						self     :   str		,     lowerCamelCase__     :   Dict		,     lowerCamelCase__     :   Optional[int]       )      ->Dict:



					'''simple docstring'''







					_UpperCAmelCase						:				Any	    =					self.split_shard_indices_by_worker(lowerCamelCase__		,     lowerCamelCase__       )
					return SparkExamplesIterable(self.df		,     partition_order=lowerCamelCase__       )






			@property
			def 							lowerCAmelCase__			(						self     :   List[Any]       )      ->int:



					'''simple docstring'''







					return len(self.partition_order       )






class 		lowerCAmelCase__	(	datasets.DatasetBuilder		):
			lowerCAmelCase       :		Dict				      =   SparkConfig




			def __init__(						self     :   Tuple		,     lowerCamelCase__     :   Optional[int]		,     lowerCamelCase__     :   str = None		,     lowerCamelCase__     :   str = None		,     **lowerCamelCase__     :   Tuple		,     )      ->Any:



					'''simple docstring'''







					import pyspark

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

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




			def 							lowerCAmelCase__			(						self     :   Dict       )      ->str:



					'''simple docstring'''







					def create_cache_and_write_probe(lowerCamelCase__     :   Dict       ):
							# 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__       )
							_UpperCAmelCase						:				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:
							_UpperCAmelCase						:				List[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 							lowerCAmelCase__			(						self     :   Any       )      ->Dict:



					'''simple docstring'''







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




			def 							lowerCAmelCase__			(						self     :   Optional[int]		,     lowerCamelCase__     :   Optional[int]       )      ->Tuple:



					'''simple docstring'''







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




			def 							lowerCAmelCase__			(						self     :   int		,     lowerCamelCase__     :   Optional[int]       )      ->str:



					'''simple docstring'''







					import pyspark

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

					_UpperCAmelCase						:				Optional[int]	    =					self.df.count()
					_UpperCAmelCase						:				List[str]	    =					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.
					_UpperCAmelCase						:				Union[str, Any]	    =					(
					    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
					)
					_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[Any]	    =					min(lowerCamelCase__		,     int(approx_total_size / max_shard_size       )       )
							_UpperCAmelCase						:				Optional[Any]	    =					self.df.repartition(lowerCamelCase__       )




			def 							lowerCAmelCase__			(						self     :   Tuple		,     lowerCamelCase__     :   Optional[int]		,     lowerCamelCase__     :   List[str]		,     lowerCamelCase__     :   Optional[int]		,     )      ->List[str]:



					'''simple docstring'''







					import pyspark

					_UpperCAmelCase						:				str	    =					ParquetWriter if file_format == "parquet" else ArrowWriter
					_UpperCAmelCase						:				Dict	    =					os.path.join(self._working_dir		,     os.path.basename(lowerCamelCase__       )       ) if self._working_dir else fpath
					_UpperCAmelCase						:				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						:				int	    =					self.config.features
					_UpperCAmelCase						:				Any	    =					self._writer_batch_size
					_UpperCAmelCase						:				Optional[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.
							_UpperCAmelCase						:				Tuple	    =					pyspark.TaskContext().taskAttemptId()
							_UpperCAmelCase						:				int	    =					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"]		,     )
							_UpperCAmelCase						:				Dict	    =					0
							_UpperCAmelCase						:				str	    =					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__		,     )
							_UpperCAmelCase						:				Tuple	    =					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:
											_UpperCAmelCase    ,			_UpperCAmelCase						:				Optional[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
											_UpperCAmelCase						:				Union[str, Any]	    =					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__		,     )
									_UpperCAmelCase						:				Any	    =					pa.Table.from_batches([batch]       )
									writer.write_table(lowerCamelCase__       )

							if writer._num_bytes > 0:
									_UpperCAmelCase    ,			_UpperCAmelCase						:				Union[str, 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(lowerCamelCase__       )       ):
											_UpperCAmelCase						:				Dict	    =					os.path.join(os.path.dirname(lowerCamelCase__       )		,     os.path.basename(lowerCamelCase__       )       )
											shutil.move(lowerCamelCase__		,     lowerCamelCase__       )

					_UpperCAmelCase						:				int	    =					(
					    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 							lowerCAmelCase__			(						self     :   Tuple		,     lowerCamelCase__     :   Tuple		,     lowerCamelCase__     :   Optional[Any] = "arrow"		,     lowerCamelCase__     :   str = None		,     lowerCamelCase__     :   Dict = None		,     **lowerCamelCase__     :   Optional[Any]		,     )      ->Union[str, Any]:



					'''simple docstring'''







					self._validate_cache_dir()

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

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

					_UpperCAmelCase						:				Tuple	    =					0
					_UpperCAmelCase						:				List[Any]	    =					0
					_UpperCAmelCase						:				Tuple	    =					0
					_UpperCAmelCase						:				str	    =					[]
					_UpperCAmelCase						:				str	    =					[]

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

					_UpperCAmelCase						:				int	    =					total_num_examples
					_UpperCAmelCase						:				List[str]	    =					total_num_bytes

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

							# use the -SSSSS-of-NNNNN pattern
							def _rename_shard(
							    lowerCamelCase__     :   Tuple		,     lowerCamelCase__     :   Dict		,     lowerCamelCase__     :   Union[str, Any]		,     ):
									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}"""       )		,     )

							_UpperCAmelCase						:				int	    =					[]
							_UpperCAmelCase						:				Tuple	    =					0
							for i in range(len(lowerCamelCase__       )       ):
									_UpperCAmelCase    ,			_UpperCAmelCase						:				List[Any]	    =					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
							_UpperCAmelCase						:				Dict	    =					0
							_UpperCAmelCase						:				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(lowerCamelCase__		,     ""       )		,     )






			def 							lowerCAmelCase__			(						self     :   Tuple		,     lowerCamelCase__     :   Optional[int]		,     )      ->Any:



					'''simple docstring'''







					return SparkExamplesIterable(self.df       )






import multiprocessing
import time

from arguments import PretokenizationArguments
from datasets import load_dataset

from transformers import AutoTokenizer, HfArgumentParser





def 					__lowerCamelCase	( UpperCamelCase__      ):







							'''simple docstring'''






							snake_case_							=    {}
							snake_case_							=    tokenizer(example['content']   ,							truncation=UpperCamelCase__      )['input_ids']
							snake_case_							=    len(example['content']      ) / len(output['input_ids']      )
							return output


_UpperCAmelCase		:   Dict										=       HfArgumentParser(PretokenizationArguments)
_UpperCAmelCase		:   List[Any]										=       parser.parse_args()
if args.num_workers is None:
					_UpperCAmelCase		:   Union[str, Any]										=       multiprocessing.cpu_count()
_UpperCAmelCase		:   int										=       AutoTokenizer.from_pretrained(args.tokenizer_dir)

_UpperCAmelCase		:   Optional[int]										=       time.time()
_UpperCAmelCase		:   List[str]										=       load_dataset(args.dataset_name, split="""train""")
print(F'''Dataset loaded in {time.time()-t_start:.2f}s''')

_UpperCAmelCase		:   Tuple										=       time.time()
_UpperCAmelCase		:   Union[str, Any]										=       ds.map(
    tokenize,
    num_proc=args.num_workers,
    remove_columns=[
        """repo_name""",
        """path""",
        """copies""",
        """size""",
        """content""",
        """license""",
        """hash""",
        """line_mean""",
        """line_max""",
        """alpha_frac""",
        """autogenerated""",
    ],
)
print(F'''Dataset tokenized in {time.time()-t_start:.2f}s''')

_UpperCAmelCase		:   Dict										=       time.time()
ds.push_to_hub(args.tokenized_data_repo)


print(F'''Data pushed to the hub in {time.time()-t_start:.2f}s''')




