Read UC delta table

[XiaohanZhangCMU]

Convert UC delta table into a single jsonl file stored in the local disk

To test the script from a training cluster, follow these steps

1. create sample UC tables on Databricks Notebook by

%sql 
CREATE TABLE random_table (
    prompts VARCHAR(255),
    responses VARCHAR(255)
);
%python
df = spark.range(100000).selectExpr("concat('Random Prompt ', cast(id as string)) as prompts", "concat('Random Response ', cast((rand() * 1000) as string)) as responses")

# Write the DataFrame to the table
df.write.mode("append").saveAsTable("random_table")

2. On your machine (local or gpu cluster)

export DATABRICKS_HOST="YOUR HOST NAME"
export DATABRICKS_TOKEN="YOUR TOKEN"

git clone https://github.com/mosaicml/llm-foundry.git
git fetch origin pull/773/head:delta_input
cd llm-foundry
pip install -e .[databricks]

There are a few arguments are required including delta_table_name (full table name catalog.schema.table), json_output_path (absolute local path that result jsonl to be written to) and cluster_id.

If your workspace has access to serverless (e.g., e2-dogfood), let cluster_id = any string, but --use_serverless True.

python scripts/data_prep/convert_delta_to_json.py --delta_table_name main.streaming.random_large_table --json_output_path /tmp/test_json1 --processes 128 --cluster_id asdfg --use_serverless True

Otherwise, you need to specify the actual cluster id. If the cluster has a DBR version 14.1.0 and above, the script uses parallelized databricks-connect, recommended to use.

python scripts/data_prep/convert_delta_to_json.py --delta_table_name main.streaming.random_large_table --json_output_path /tmp/test_json1 --processes 128 --cluster_id 1214-001856-19o83v16

For clusters with DBR less than 14.1.0, the script will use sql-connect. In this scenario, the cluster's http_path is required. Note: if you did provide a http_path, the script will default to use sql-connect regardless the cluster's DBR version.

python scripts/data_prep/convert_delta_to_json.py --delta_table_name main.streaming.random_large_table --json_output_path /tmp/test_json1 --processes 128 --cluster_id 0110-003946-ababdevs --http_path sql/protocolv1/o/6051921418418893/0110-003946-ababdevs

The minimum version required is 12.2.0, any cluster with DBR version lower than that will error out because "deletionVectors" feature is not available.

Performance and known issues

Serverless

table	nrows	json size	n threads	download (s)
main.streaming.dummy_table	12	4.0K	128	3.6
main.streaming.random_table	100K	8.3M	128	4.1
main.streaming.wikipedia	5M	33G	128	165

With dedicated clusters (policy needs to be shared), some runtime does not support.

cluster-id	runtime version	ok/fail runtime/Err
"1214-001856-19o83v16"	14.1	ok/8s
"1214-001919-ugt7zlm"	14.2	ok/7.2s.
"1214-001836-tnyy6g0b"	14.0	fail/Err-1.
"1214-001808-x2a8jyst"	13.3LTS	fail/Err-2
"1214-001936-oe0wtegk"	15	ok/8.6s

Err-1:

An error occurred: Expecting value: line 1 column 1 (char 0)
An error occurred: Expecting value: line 1 column 1 (char 0)

Err-2:

status = StatusCode.UNIMPLEMENTED
details = "Method not found: spark.connect.SparkConnectService/ReleaseExecute"
debug_error_string = "UNKNOWN:Error received from peer  {created_time:"2023-12-14T07:50:24.764993328+00:00", grpc_status:12, grpc_message:"Method not found: spark.connect.SparkConnectService/ReleaseExecute"}"

Dependency:
After this PR is merged. The script needs to be refactored to remove the patch. https://github.com/databricks/runtime/pull/79131/files

[panchalhp-db]

LGTM!

[nancyhung]

A few non-blocking comments! Could you link your reference DB notebook you're iterating with on the PR description for reference?