docstr: Parallelized implementation of `log_artifacts`.
docstr_tok: ["Parallelized","implementation","of","`","log_artifacts","`","."]
code: def log_artifacts(self, local_dir, artifact_path=None):
      """
      Parallelized implementation of `log_artifacts`.
      """
  
      artifact_path = artifact_path or ""
  
      staged_uploads = []
      for dirpath, _, filenames in os.walk(local_dir):
          artifact_subdir = artifact_path
          if dirpath != local_dir:
              rel_path = os.path.relpath(dirpath, local_dir)
              rel_path = relative_path_to_artifact_path(rel_path)
              artifact_subdir = posixpath.join(artifact_path, rel_path)
          for name in filenames:
              src_file_path = os.path.join(dirpath, name)
              src_file_name = os.path.basename(src_file_path)
              staged_uploads.append(
                  StagedArtifactUpload(
                      src_file_path=src_file_path,
                      artifact_file_path=posixpath.join(
                          artifact_subdir, src_file_name
                      ),
                  )
              )
  
      # Join futures to ensure that all artifacts have been uploaded prior to returning
      failed_uploads = {}
  
      # For each batch of files, upload them in parallel and wait for completion
      # TODO: change to class method
      def upload_artifacts_iter():
          for staged_upload_chunk in chunk_list(
              staged_uploads, _ARTIFACT_UPLOAD_BATCH_SIZE
          ):
              write_credential_infos = self._get_write_credential_infos(
                  remote_file_paths=[
                      staged_upload.artifact_file_path
                      for staged_upload in staged_upload_chunk
                  ],
              )
  
              inflight_uploads = {}
              for staged_upload, write_credential_info in zip(
                  staged_upload_chunk, write_credential_infos
              ):
                  upload_future = self.thread_pool.submit(
                      self._upload_to_cloud,
                      cloud_credential_info=write_credential_info,
                      src_file_path=staged_upload.src_file_path,
                      artifact_file_path=staged_upload.artifact_file_path,
                  )
                  inflight_uploads[
                      staged_upload.src_file_path
                  ] = upload_future
  
              yield from inflight_uploads.items()
  
      with ArtifactProgressBar.files(
          desc="Uploading artifacts", total=len(staged_uploads)
      ) as pbar:
          if len(staged_uploads) >= 10 and pbar.pbar:
              _logger.info(
                  "The progress bar can be disabled by setting the environment "
                  f"variable {MLFLOW_ENABLE_ARTIFACTS_PROGRESS_BAR} to false"
              )
          for src_file_path, upload_future in upload_artifacts_iter():
              try:
                  upload_future.result()
                  pbar.update()
              except Exception as e:
                  failed_uploads[src_file_path] = repr(e)
  
      if len(failed_uploads) > 0:
          raise MlflowException(
              message=(
                  "The following failures occurred while uploading one or more artifacts"
                  f" to {self.artifact_uri}: {failed_uploads}"
              )
          )
  
imports: import logging
  import math
  import os
  import posixpath
  from abc import abstractmethod
  from collections import namedtuple
  from concurrent.futures import as_completed
  from mlflow.environment_variables import MLFLOW_ENABLE_ARTIFACTS_PROGRESS_BAR, MLFLOW_ENABLE_MULTIPART_DOWNLOAD, MLFLOW_MULTIPART_DOWNLOAD_CHUNK_SIZE, MLFLOW_MULTIPART_UPLOAD_CHUNK_SIZE, MLFLOW_MULTIPART_UPLOAD_MINIMUM_FILE_SIZE
  from mlflow.exceptions import MlflowException
  from mlflow.store.artifact.artifact_repo import ArtifactRepository
  from mlflow.utils import chunk_list
  from mlflow.utils.file_utils import ArtifactProgressBar, download_chunk_retries, parallelized_download_file_using_http_uri, relative_path_to_artifact_path, remove_on_error
  from mlflow.utils.uri import is_fuse_or_uc_volumes_uri
label_desc: Use image node_id 2 for calling the CloudArtifactRepository obj's underlying member method code with example usage: obj.log_artifacts(local_dir, artifact_path) without return types