class_docstr: Parquet datasource, for reading and writing Parquet files.
  
  The primary difference from ParquetBaseDatasource is that this uses
  PyArrow's `ParquetDataset` abstraction for dataset reads, and thus offers
  automatic Arrow dataset schema inference and row count collection at the
  cost of some potential performance and/or compatibility penalties.
class_docstr_tok: ["Parquet","datasource",",","for","reading","and","writing","Parquet","files",".","The","primary","difference","from","ParquetBaseDatasource","is","that","this","uses","PyArrow","'s","`","ParquetDataset","`","abstraction","for","dataset","reads",",","and","thus","offers","automatic","Arrow","dataset","schema","inference","and","row","count","collection","at","the","cost","of","some","potential","performance","and\/or","compatibility","penalties","."]
code: def __init__(
      self,
      paths: Union[str, List[str]],
      *,
      columns: Optional[List[str]] = None,
      dataset_kwargs: Optional[Dict[str, Any]] = None,
      to_batch_kwargs: Optional[Dict[str, Any]] = None,
      _block_udf: Optional[Callable[[Block], Block]] = None,
      filesystem: Optional["pyarrow.fs.FileSystem"] = None,
      schema: Optional[Union[type, "pyarrow.lib.Schema"]] = None,
      meta_provider: ParquetMetadataProvider = DefaultParquetMetadataProvider(),
      partition_filter: PathPartitionFilter = None,
      shuffle: Union[Literal["files"], None] = None,
      include_paths: bool = False,
      file_extensions: Optional[List[str]] = None,
  ):
      _check_pyarrow_version()
  
      import pyarrow as pa
      import pyarrow.parquet as pq
  
      self._supports_distributed_reads = not _is_local_scheme(paths)
      if (
          not self._supports_distributed_reads
          and ray.util.client.ray.is_connected()
      ):
          raise ValueError(
              "Because you're using Ray Client, read tasks scheduled on the Ray "
              "cluster can't access your local files. To fix this issue, store "
              "files in cloud storage or a distributed filesystem like NFS."
          )
  
      self._local_scheduling = None
      if not self._supports_distributed_reads:
          from ray.util.scheduling_strategies import (
              NodeAffinitySchedulingStrategy,
          )
  
          self._local_scheduling = NodeAffinitySchedulingStrategy(
              ray.get_runtime_context().get_node_id(), soft=False
          )
  
      paths, filesystem = _resolve_paths_and_filesystem(
          paths, filesystem
      )
  
      # HACK: PyArrow's `ParquetDataset` errors if input paths contain non-parquet
      # files. To avoid this, we expand the input paths with the default metadata
      # provider and then apply the partition filter or file extensions.
      if partition_filter is not None or file_extensions is not None:
          default_meta_provider = get_generic_metadata_provider(
              file_extensions=None
          )
          expanded_paths, _ = map(
              list,
              zip(
                  *default_meta_provider.expand_paths(paths, filesystem)
              ),
          )
  
          paths = list(expanded_paths)
          if partition_filter is not None:
              paths = partition_filter(paths)
          if file_extensions is not None:
              paths = [
                  path
                  for path in paths
                  if _has_file_extension(path, file_extensions)
              ]
  
          filtered_paths = set(expanded_paths) - set(paths)
          if filtered_paths:
              logger.info(f"Filtered out {len(filtered_paths)} paths")
      else:
          if len(paths) == 1:
              paths = paths[0]
  
      if dataset_kwargs is None:
          dataset_kwargs = {}
  
      try:
          pq_ds = pq.ParquetDataset(
              paths,
              **dataset_kwargs,
              filesystem=filesystem,
              use_legacy_dataset=False,
          )
      except OSError as e:
          _handle_read_os_error(e, paths)
      if schema is None:
          schema = pq_ds.schema
      if columns:
          schema = pa.schema(
              [schema.field(column) for column in columns],
              schema.metadata,
          )
  
      if _block_udf is not None:
          # Try to infer dataset schema by passing dummy table through UDF.
          dummy_table = schema.empty_table()
          try:
              inferred_schema = _block_udf(dummy_table).schema
              inferred_schema = inferred_schema.with_metadata(
                  schema.metadata
              )
          except Exception:
              logger.debug(
                  "Failed to infer schema of dataset by passing dummy table "
                  "through UDF due to the following exception:",
                  exc_info=True,
              )
              inferred_schema = schema
      else:
          inferred_schema = schema
  
      try:
          prefetch_remote_args = {}
          prefetch_remote_args[
              "num_cpus"
          ] = NUM_CPUS_FOR_META_FETCH_TASK
          if self._local_scheduling:
              prefetch_remote_args[
                  "scheduling_strategy"
              ] = self._local_scheduling
          else:
              # Use the scheduling strategy ("SPREAD" by default) provided in
              # `DataContext``, to spread out prefetch tasks in cluster, avoid
              # AWS S3 throttling error.
              # Note: this is the same scheduling strategy used by read tasks.
              prefetch_remote_args[
                  "scheduling_strategy"
              ] = DataContext.get_current().scheduling_strategy
          if RETRY_EXCEPTIONS_FOR_META_FETCH_TASK is not None:
              prefetch_remote_args[
                  "retry_exceptions"
              ] = RETRY_EXCEPTIONS_FOR_META_FETCH_TASK
  
          self._metadata = (
              meta_provider.prefetch_file_metadata(
                  pq_ds.fragments, **prefetch_remote_args
              )
              or []
          )
      except OSError as e:
          _handle_read_os_error(e, paths)
  
      if to_batch_kwargs is None:
          to_batch_kwargs = {}
  
      # NOTE: Store the custom serialized `ParquetFileFragment` to avoid unexpected
      # network calls when `_ParquetDatasourceReader` is serialized. See
      # `_SerializedFragment()` implementation for more details.
      self._pq_fragments = [
          _SerializedFragment(p) for p in pq_ds.fragments
      ]
      self._pq_paths = [p.path for p in pq_ds.fragments]
      self._meta_provider = meta_provider
      self._inferred_schema = inferred_schema
      self._block_udf = _block_udf
      self._to_batches_kwargs = to_batch_kwargs
      self._columns = columns
      self._schema = schema
      self._encoding_ratio = self._estimate_files_encoding_ratio()
      self._file_metadata_shuffler = None
      self._include_paths = include_paths
      if shuffle == "files":
          self._file_metadata_shuffler = np.random.default_rng()
  
imports: import logging
  from typing import TYPE_CHECKING, Any, Callable, Dict, Iterator, List, Literal, Optional, Union
  import numpy
  import ray
  import ray.cloudpickle
  from ray.data._internal.progress_bar import ProgressBar
  from ray.data._internal.remote_fn import cached_remote_fn
  from ray.data._internal.util import _check_pyarrow_version, _is_local_scheme
  from ray.data.block import Block
  from ray.data.context import DataContext
  from ray.data.datasource import Datasource
  from ray.data.datasource._default_metadata_providers import get_generic_metadata_provider
  from ray.data.datasource.datasource import ReadTask
  from ray.data.datasource.file_meta_provider import DefaultParquetMetadataProvider, ParquetMetadataProvider, _handle_read_os_error
  from ray.data.datasource.partitioning import PathPartitionFilter
  from ray.data.datasource.path_util import _has_file_extension, _resolve_paths_and_filesystem
  from ray.util.annotations import PublicAPI
label_desc: Use image node_id 1 to create a new ParquetDatasource object from inherited base classes: Datasource with example: obj = ParquetDatasource(paths)