def fetch_DT(
    delta_table_name: str,
    json_output_folder: str,
    http_path: Optional[str],
    cluster_id: Optional[str],
    use_serverless: bool,
    DATABRICKS_HOST: str,
    DATABRICKS_TOKEN: str,
    batch_size: int = 1 << 30,
    processes: int = os.cpu_count(), # type: ignore
    json_output_filename: str = 'train-00000-of-00001.jsonl',
) -> None:
    """Fetch UC Delta Table to local as jsonl."""
    log.info(f'Start .... Convert delta to json')

    obj = urllib.parse.urlparse(json_output_folder)
    if obj.scheme != '':
        raise ValueError(
            'Check the json_output_folder and verify it is a local path!',
        )

    if os.path.exists(json_output_folder):
        if not os.path.isdir(json_output_folder) or os.listdir(
            json_output_folder,
        ):
            raise RuntimeError(
                f'Output folder {json_output_folder} already exists and is not empty. Please remove it and retry.',
            )

    os.makedirs(json_output_folder, exist_ok=True)

    if not json_output_filename.endswith('.jsonl'):
        raise ValueError('json_output_filename needs to be a jsonl file')

    log.info(f'Directory {json_output_folder} created.')

    # Validate_and_get_cluster_info allows cluster_id to be None if use_serverless is True.
    method, dbsql, sparkSession = validate_and_get_cluster_info(
        cluster_id=cluster_id,
        databricks_host=DATABRICKS_HOST,
        databricks_token=DATABRICKS_TOKEN,
        http_path=http_path,
        use_serverless=use_serverless,
    )

    formatted_delta_table_name = format_tablename(delta_table_name)
    import grpc
    import pyspark.errors.exceptions.connect as spark_errors
    try:
        fetch(
            method,
            formatted_delta_table_name,
            json_output_folder,
            batch_size,
            processes,
            sparkSession,
            dbsql,
        )
    except (grpc.RpcError, spark_errors.SparkConnectGrpcException) as e:
        if isinstance(
            e,
            spark_errors.SparkConnectGrpcException,
        ) and 'is not Shared or Single User Cluster' in str(e):
            raise FaultyDataPrepCluster(
                message=
                f'The cluster you have provided: {cluster_id} does not have data governance enabled. Please use a cluster with a data security mode other than NONE. {e}',
            ) from e
        if isinstance(
            e,
            spark_errors.SparkConnectGrpcException,
        ) and 'Cannot start cluster' in str(e):
            raise FaultyDataPrepCluster(
                message=
                f'The data preparation cluster you provided is terminated. Please retry with a cluster that is healthy and alive. {e}',
            ) from e
        if isinstance(
            e,
            spark_errors.SparkConnectGrpcException,
        ) and 'is not usable' in str(e):
            raise FaultyDataPrepCluster(
                message=
                f'The data preparation cluster you provided is not usable. Please retry with a cluster that is healthy and alive. {e}',
            ) from e
        if isinstance(
            e,
            spark_errors.SparkConnectGrpcException,
        ) and 'do not have permission to attach to cluster' in str(e):
            raise FaultyDataPrepCluster(
                message=
                f'You do not have permission to attach to the data preparation cluster you provided. {e}',
            ) from e
        if isinstance(e, grpc.RpcError) and e.code(
        ) == grpc.StatusCode.INTERNAL and 'Job aborted due to stage failure' in e.details(
        ):
            raise FaultyDataPrepCluster(
                message=
                f'Faulty data prep cluster, please try swapping data prep cluster: {e.details()}',
            ) from e
        raise e

    if dbsql is not None:
        dbsql.close()

    # Combine downloaded jsonl into one big jsonl for IFT.
    iterative_combine_jsons(
        json_output_folder,
        os.path.join(json_output_folder, json_output_filename),
    )

    _validate_written_file(
        json_output_folder,
        json_output_filename,
        delta_table_name,
    )
Fetch UC Delta Table to local as jsonl.
llmfoundry/command_utils/data_prep/convert_delta_to_json.py