class_docstr: Shuffles and writes Examples to Arrow files.
  The Arrow files are converted from Parquet files that are the output of Apache Beam pipelines.
class_docstr_tok: ["Shuffles","and","writes","Examples","to","Arrow","files",".","The","Arrow","files","are","converted","from","Parquet","files","that","are","the","output","of","Apache","Beam","pipelines","."]
docstr: Run after the pipeline has finished.
  It converts the resulting parquet files to arrow and it completes the info from the pipeline metrics.
  
  Args:
      metrics_query_result: `dict` obtained from pipeline_results.metrics().query(m_filter). Make sure
          that the filter keeps only the metrics for the considered split, under the namespace `split_name`.
docstr_tok: ["Run","after","the","pipeline","has","finished",".","It","converts","the","resulting","parquet","files","to","arrow","and","it","completes","the","info","from","the","pipeline","metrics",".","Args",":","metrics_query_result",":","`","dict","`","obtained","from","pipeline_results.metrics","(",")",".query","(","m_filter",")",".","Make","sure","that","the","filter","keeps","only","the","metrics","for","the","considered","split",",","under","the","namespace","`","split_name","`","."]
code: def finalize(self, metrics_query_result: dict):
      """
      Run after the pipeline has finished.
      It converts the resulting parquet files to arrow and it completes the info from the pipeline metrics.
  
      Args:
          metrics_query_result: `dict` obtained from pipeline_results.metrics().query(m_filter). Make sure
              that the filter keeps only the metrics for the considered split, under the namespace `split_name`.
      """
      import apache_beam as beam
  
      from .utils import beam_utils
  
      # Beam FileSystems require the system's path separator in the older versions
      fs, _, [parquet_path] = fsspec.get_fs_token_paths(
          self._parquet_path
      )
      parquet_path = (
          str(Path(parquet_path))
          if not is_remote_filesystem(fs)
          else fs.unstrip_protocol(parquet_path)
      )
  
      shards_metadata = list(
          beam.io.filesystems.FileSystems.match(
              [parquet_path + "*.parquet"]
          )[0].metadata_list
      )
      shards = [metadata.path for metadata in shards_metadata]
      num_bytes = sum(
          [metadata.size_in_bytes for metadata in shards_metadata]
      )
      shard_lengths = get_parquet_lengths(shards)
  
      # Convert to arrow
      if self._path.endswith(".arrow"):
          logger.info(
              f"Converting parquet files {self._parquet_path} to arrow {self._path}"
          )
          shards = [
              metadata.path
              for metadata in beam.io.filesystems.FileSystems.match(
                  [parquet_path + "*.parquet"]
              )[0].metadata_list
          ]
          try:  # stream conversion
              num_bytes = 0
              for shard in hf_tqdm(shards, unit="shards"):
                  with beam.io.filesystems.FileSystems.open(
                      shard
                  ) as source:
                      with beam.io.filesystems.FileSystems.create(
                          shard.replace(".parquet", ".arrow")
                      ) as destination:
                          shard_num_bytes, _ = parquet_to_arrow(
                              source, destination
                          )
                          num_bytes += shard_num_bytes
          except (
              OSError
          ) as e:  # broken pipe can happen if the connection is unstable, do local conversion instead
              if e.errno != errno.EPIPE:  # not a broken pipe
                  raise
              logger.warning(
                  "Broken Pipe during stream conversion from parquet to arrow. Using local convert instead"
              )
              local_convert_dir = os.path.join(
                  self._cache_dir, "beam_convert"
              )
              os.makedirs(local_convert_dir, exist_ok=True)
              num_bytes = 0
              for shard in hf_tqdm(shards, unit="shards"):
                  local_parquet_path = os.path.join(
                      local_convert_dir,
                      hash_url_to_filename(shard) + ".parquet",
                  )
                  beam_utils.download_remote_to_local(
                      shard, local_parquet_path
                  )
                  local_arrow_path = local_parquet_path.replace(
                      ".parquet", ".arrow"
                  )
                  shard_num_bytes, _ = parquet_to_arrow(
                      local_parquet_path, local_arrow_path
                  )
                  num_bytes += shard_num_bytes
                  remote_arrow_path = shard.replace(
                      ".parquet", ".arrow"
                  )
                  beam_utils.upload_local_to_remote(
                      local_arrow_path, remote_arrow_path
                  )
  
      # Save metrics
      counters_dict = {
          metric.key.metric.name: metric.result
          for metric in metrics_query_result["counters"]
      }
      self._num_examples = counters_dict["num_examples"]
      self._num_bytes = num_bytes
      self._shard_lengths = shard_lengths
      return self._num_examples, self._num_bytes
  
imports: import errno
  import json
  import os
  import sys
  from pathlib import Path
  from typing import Any, Dict, Iterable, List, Optional, Tuple, Union
  import fsspec
  import numpy
  import pyarrow
  import pyarrow.parquet
  from .None import config
  from .features import Features, Image, Value
  from .features.features import FeatureType, _ArrayXDExtensionType, cast_to_python_objects, generate_from_arrow_type, get_nested_type, list_of_np_array_to_pyarrow_listarray, numpy_to_pyarrow_listarray, to_pyarrow_listarray
  from .filesystems import is_remote_filesystem
  from .info import DatasetInfo
  from .keyhash import DuplicatedKeysError, KeyHasher
  from .table import array_cast, array_concat, cast_array_to_feature, embed_table_storage, table_cast
  from .utils import logging
  from .utils import tqdm
  from .utils.file_utils import hash_url_to_filename
  from .utils.py_utils import asdict, first_non_null_value
label_desc: Use image node_id 3 for calling the BeamWriter obj's underlying member method code with example usage: obj.finalize(metrics_query_result) and returns: self,  self