code: def _service_vep(
      backend: ServiceBackend,
      ht: Table,
      config: Optional[VEPConfig],
      block_size: int,
      csq: bool,
      tolerate_parse_error: bool,
      temp_input_directory: str,
      temp_output_directory: str,
  ) -> hl.Table:
      reference_genome = ht.locus.dtype.reference_genome.name
      cloud = async_to_blocking(backend._batch_client.cloud())
      regions = backend.regions
  
      if config is not None:
          vep_config = config
      else:
          vep_config = _supported_vep_config(
              cloud, reference_genome, regions=regions
          )
  
      requester_pays_project = backend.flags.get(
          "gcs_requester_pays_project"
      )
      if (
          requester_pays_project is None
          and vep_config.data_bucket_is_requester_pays
          and vep_config.cloud == "gcp"
      ):
          raise ValueError(
              "No requester pays project has been set. "
              "Use hl.init(gcs_requester_pays_configuration='MY_PROJECT') "
              "to set the requester pays project to use."
          )
  
      if csq:
          vep_typ = hl.tarray(hl.tstr)
      else:
          vep_typ = vep_config.json_typ
  
      def build_vep_batch(
          b: bc.aioclient.Batch,
          vep_input_path: str,
          vep_output_path: str,
      ):
          if csq:
              local_output_file = "/io/output"
              vep_command = vep_config.command(
                  consequence=csq,
                  part_id=-1,
                  input_file=None,
                  output_file=local_output_file,
                  tolerate_parse_error=tolerate_parse_error,
              )
              env = {
                  "VEP_BLOCK_SIZE": str(block_size),
                  "VEP_DATA_MOUNT": shq(vep_config.data_mount),
                  "VEP_CONSEQUENCE": str(int(csq)),
                  "VEP_TOLERATE_PARSE_ERROR": str(
                      int(tolerate_parse_error)
                  ),
                  "VEP_PART_ID": str(-1),
                  "VEP_OUTPUT_FILE": local_output_file,
                  "VEP_COMMAND": vep_command,
              }
              env.update(vep_config.env)
              b.create_job(
                  vep_config.image,
                  vep_config.batch_run_csq_header_command,
                  attributes={"name": "csq-header"},
                  resources={"cpu": "1", "memory": "standard"},
                  cloudfuse=[
                      (
                          vep_config.data_bucket,
                          vep_config.data_mount,
                          True,
                      )
                  ],
                  output_files=[
                      (
                          local_output_file,
                          f"{vep_output_path}/csq-header",
                      )
                  ],
                  regions=vep_config.regions,
                  requester_pays_project=requester_pays_project,
                  env=env,
              )
  
          for f in hl.hadoop_ls(vep_input_path):
              path = f["path"]
              part_name = os.path.basename(path)
              if not part_name.startswith("part-"):
                  continue
              part_id = int(part_name.split("-")[1])
  
              local_input_file = "/io/input"
              local_output_file = "/io/output.gz"
  
              vep_command = vep_config.command(
                  consequence=csq,
                  part_id=part_id,
                  input_file=local_input_file,
                  output_file=local_output_file,
                  tolerate_parse_error=tolerate_parse_error,
              )
  
              env = {
                  "VEP_BLOCK_SIZE": str(block_size),
                  "VEP_DATA_MOUNT": shq(vep_config.data_mount),
                  "VEP_CONSEQUENCE": str(int(csq)),
                  "VEP_TOLERATE_PARSE_ERROR": str(
                      int(tolerate_parse_error)
                  ),
                  "VEP_PART_ID": str(-1),
                  "VEP_INPUT_FILE": local_input_file,
                  "VEP_OUTPUT_FILE": local_output_file,
                  "VEP_COMMAND": vep_command,
              }
              env.update(vep_config.env)
  
              b.create_job(
                  vep_config.image,
                  vep_config.batch_run_command,
                  attributes={"name": f"vep-{part_id}"},
                  resources={"cpu": "1", "memory": "standard"},
                  input_files=[(path, local_input_file)],
                  output_files=[
                      (
                          local_output_file,
                          f"{vep_output_path}/annotations/{part_name}.tsv.gz",
                      )
                  ],
                  cloudfuse=[
                      (
                          vep_config.data_bucket,
                          vep_config.data_mount,
                          True,
                      )
                  ],
                  regions=vep_config.regions,
                  requester_pays_project=requester_pays_project,
                  env=env,
              )
  
      hl.export_vcf(
          ht, temp_input_directory, parallel="header_per_shard"
      )
  
      starting_job_id = (
          async_to_blocking(backend._batch.status())["n_jobs"] + 1
      )
  
      b = bc.client.Batch(backend._batch)
      build_vep_batch(b, temp_input_directory, temp_output_directory)
  
      b.submit(disable_progress_bar=True)
  
      try:
          status = b.wait(
              description="vep(...)",
              disable_progress_bar=backend.disable_progress_bar,
              progress=None,
              starting_job=starting_job_id,
          )
      except BaseException as e:
          if isinstance(e, KeyboardInterrupt):
              print(
                  "Received a keyboard interrupt, cancelling the batch..."
              )
          b.cancel()
          backend._batch = None
          raise
  
      if status["n_succeeded"] != status["n_jobs"]:
          failing_job = [job for job in b.jobs("!success")][0]
          failing_job = b.get_job(failing_job["job_id"])
          message = {
              "batch_status": status,
              "job_status": failing_job.status(),
              "log": failing_job.log(),
          }
          raise FatalError(yamlx.dump(message))
  
      annotations = hl.import_table(
          f"{temp_output_directory}/annotations/*",
          types={
              "variant": hl.tstr,
              "vep": vep_typ,
              "part_id": hl.tint,
              "block_id": hl.tint,
          },
          force=True,
      )
  
      annotations = annotations.annotate(
          vep_proc_id=hl.struct(
              part_id=annotations.part_id, block_id=annotations.block_id
          )
      )
      annotations = annotations.drop("part_id", "block_id")
      annotations = annotations.key_by(
          **hl.parse_variant(
              annotations.variant, reference_genome=reference_genome
          )
      )
      annotations = annotations.drop("variant")
  
      if csq:
          with hl.hadoop_open(
              f"{temp_output_directory}/csq-header"
          ) as f:
              vep_csq_header = f.read().rstrip()
          annotations = annotations.annotate_globals(
              vep_csq_header=vep_csq_header
          )
  
      return annotations
  
imports: import abc
  import logging
  import hail
  from collections import Counter
  import os
  from shlex import quote
  from typing import Dict, Tuple, List, Optional, Union
  from hailtop import pip_version
  from hailtop.utils import async_to_blocking
  import hailtop.batch_client
  from hailtop.config import get_deploy_config
  from hailtop import yamlx
  from hail.backend.service_backend import ServiceBackend
  from hail.typecheck import typecheck, oneof, anytype, nullable, numeric
  from hail.expr.expressions.expression_typecheck import expr_float64
  from hail.utils import FatalError
  from hail.utils.java import Env, info, warning
  from hail.utils.misc import divide_null, guess_cloud_spark_provider, new_temp_file
  from hail.matrixtable import MatrixTable
  from hail.table import Table
  from hail.ir import TableToTableApply
  from .misc import require_biallelic, require_row_key_variant, require_col_key_str, require_table_key_variant, require_alleles_field
label_desc: Use image node_id 5 for calling a global function with example usage: _service_vep(backend, ht, config, block_size, csq, tolerate_parse_error, temp_input_directory, temp_output_directory) and returns: annotations