batch loading support

[nixxholas]

Can't seem to find documentation relating to this. Are we missing this feature or am I unable to spot the feature in the Job struct?

Was thinking if we can support batch loading with this client. That would be perfect and extremely economical.
Cool Example Here

[lquerel]

I'd like to understand your use case a little better. Do you want to load local files into BigQuery or do you want to load files stored in GCS into BigQuery? What is the initial format of your files?

[nixxholas]

What I'm thinking would be this:

https://cloud.google.com/bigquery/docs/loading-data-cloud-storage-json#loading_json_data_into_a_new_table

using Google.Apis.Bigquery.v2.Data;
using Google.Cloud.BigQuery.V2;
using System;

public class BigQueryLoadTableGcsJson
{
    public void LoadTableGcsJson(
        string projectId = "your-project-id",
        string datasetId = "your_dataset_id"
    )
    {
        BigQueryClient client = BigQueryClient.Create(projectId);
        var gcsURI = "gs://cloud-samples-data/bigquery/us-states/us-states.json";
        var dataset = client.GetDataset(datasetId);
        var schema = new TableSchemaBuilder {
            { "name", BigQueryDbType.String },
            { "post_abbr", BigQueryDbType.String }
        }.Build();
        TableReference destinationTableRef = dataset.GetTableReference(
            tableId: "us_states");
        // Create job configuration
        var jobOptions = new CreateLoadJobOptions()
        {
            SourceFormat = FileFormat.NewlineDelimitedJson
        };
        // Create and run job
        BigQueryJob loadJob = client.CreateLoadJob(
            sourceUri: gcsURI, destination: destinationTableRef,
            schema: schema, options: jobOptions);
        loadJob.PollUntilCompleted();  // Waits for the job to complete.
        // Display the number of rows uploaded
        BigQueryTable table = client.GetTable(destinationTableRef);
        Console.WriteLine(
            $"Loaded {table.Resource.NumRows} rows to {table.FullyQualifiedId}");
    }
}

[nixxholas]

Simply put, maybe for the first phase of implementation, we'll explicitly support JSON, and perform a LoadJob op similar to the code sample above provided by Github.

Kind of like

1. User initialises a LoadJob
2. User pushes an object / Vec of objects into the Job
3. User pushes the Job to one of our APIs.
4. API prepares the dataset into whatever is appropriate.

Can we have 0 files involved (everything in-mem, similar to how streaming inserts would behave)

I'm also assuming that the GCS involved in this context would be the one directly used by the BQ in question.

[lquerel]

I'm currently on vacation, I'll look into it as soon as possible. The job.insert method can probably be used for that. I need to look at it more precisely.

[nixxholas]

@lquerel Enjoy your vacation! I'll explore job.insert on my own and aid in the documentation further if I get through!

[nixxholas]

Are we interested in covering all capabilities of BQ? If so, we may have to depend on this https://github.com/ThouCheese/cloud-storage-rs

I'm close to a working code example.

[lquerel]

Yes, covering all the capabilities of BQ is the goal, even if many things are still missing (e.g. the storage read/write API). I would also like to minimize dependencies as much as possible. So adding a direct dependency to cloud-storage-rs is not ideal but with a little refactoring we could probably create an example with this dependency and have the BQ sdk take the link generated by cloud-storage-rs as a parameter.
At this point, I suggest you add this dependency to finalize a 1st example and I'll do the necessary to externalize the example in a sub-crate to not pollute the main crate.

[nixxholas]

This is the updated broken sample along with the GCS crate.

const CHARSET: &[u8] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789";

pub async fn batch_load(
    borrowed_app_state: &AppState,
    client: gcp_bigquery_client::Client,
    table_id: &str,
    data: &[u8],
    dry_run: Option<bool>,
) -> bool {
    // Obtain the rng machine to uniquely generate a random name for the temp file.
    const CHARSET_LEN: usize = 30;
    let mut rng = rand::thread_rng();

    let rng_file_name: String = (0..CHARSET_LEN)
        .map(|_| {
            let idx = rng.gen_range(0..CHARSET.len());
            CHARSET[idx] as char
        })
        .collect();
    // Create and push the data to GCS
    let object_res = Object::create(
        &borrowed_app_state.gcs_bucket_name,
        data,
        &rng_file_name,
        "application/text",
    );

    if let Ok(object) = object_res {
        // Push the GCS file to BQ
        let table_ref = TableReference::new(
            &borrowed_app_state.gcp_project_id,
            &borrowed_app_state.gcp_dataset_id,
            &table_id,
        );
        let job = Job {
            configuration: Some(JobConfiguration {
                copy: None,
                dry_run,
                extract: None,
                job_timeout_ms: Some("30000".to_string()),
                job_type: None,
                labels: None,
                load: Some(JobConfigurationLoad {
                    allow_jagged_rows: Some(true),
                    allow_quoted_newlines: None,
                    autodetect: Some(true),
                    clustering: None,
                    // Never allow this job to create new tables.
                    create_disposition: Some("CREATE_NEVER".to_string()),
                    decimal_target_types: None,
                    destination_encryption_configuration: None,
                    destination_table: Some(table_ref),
                    destination_table_properties: None,
                    encoding: None,
                    field_delimiter: None,
                    hive_partitioning_options: None,
                    ignore_unknown_values: None,
                    // Default to JSON for now
                    json_extension: None,
                    // None = no bad records are allowed.
                    max_bad_records: None,
                    null_marker: None,
                    projection_fields: None,
                    quote: None,
                    range_partitioning: None,
                    schema: None,
                    schema_inline: None,
                    schema_inline_format: None,
                    schema_update_options: None,
                    skip_leading_rows: None,
                    source_format: Some("NEWLINE_DELIMITED_JSON".to_string()),
                    source_uris: Some(vec![object.download_url(100000).unwrap()]),
                    time_partitioning: None,
                    use_avro_logical_types: None,
                    write_disposition: None,
                }),
                query: None,
            }),
            job_reference: Some(JobReference {
                job_id: Some(rng_file_name),
                // location: Some("us-east1".to_string()),
                location: None,
                project_id: Some(borrowed_app_state.gcp_project_id.clone()),
            }),
            statistics: None,
            status: None,
            etag: None,
            id: None,
            kind: None,
            self_link: None,
            user_email: None,
        };

        let client_result = client
            .job()
            .insert(&borrowed_app_state.gcp_project_id, job)
            .await;

        if let Ok(job) = client_result {
            let job_status = job.status.unwrap();
            if job_status.state.unwrap() != "DONE" {
                let reason = if let Some(job_err) = job_status.error_result {
                    if let Some(msg) = job_err.reason {
                        msg
                    } else {
                        "An unknown job error has occurred.".to_string()
                    }
                } else {
                    "An unknown job failure for BQ insertion has occurred.".to_string()
                };
                error!("{}", reason);

                return false;
            }

            return true;
        } else {
            let wrapped_err = client_result.err();

            if let Some(err) = wrapped_err {
                if let Some(msg) = get_error_if_exists(err) {
                    error!("{}", msg);
                    false
                } else {
                    error!(
                        "[batch_loading] A fatal, unknown error occurred when pushing the \
                    GCS object to BQ."
                    );
                    false
                }
            } else {
                error!("An unknown error has occurred while batch loading rows to BQ.");
                false
            }
        }
    } else {
        error!("Failed to ingest block into GCS.");
        false
    }
}

[nixxholas]

Job is still RUNNING even though it was await-ed..

[lquerel]

Hi Nicholas - I will take a look at it this weekend.

[lquerel]

Although the existing API could be simplified, you can implement what you need by following the example below.

https://github.com/lquerel/gcp-bigquery-client/blob/main/examples/bq_load_job.rs

The issue was not in the client itself but in the initial example. Instead of passing a download url, the JobConfigurationLoad object was expecting an URI (gs:///.

Let me know if everything is working now for you.

[nixxholas]

Hi @lquerel!

I fixed everything on my own during the weekends. Everything seem to work fine, except for the part where i want to constant wait until the job status is complete or has failed.

I was wondering should we ingest the example functions into the library? Could help a lot!

[lquerel]

Yes I plan to add these functions in the library directly, probably for the next version.

What is the issue with the status? I have a loop over the get_job endpoint to check the status and wait until DONE and it works fine.

[nixxholas]

It still wouldn't cover edge cases as intended.

Request error (error: builder error: unsupported pair)

sometimes occurs even though it has already successfully been pushed to BQ.

[nixxholas]

I can confirm this is reproducible, with your same code. Is this an environment issue?

ARM64 M1.

[nixxholas]

The difference between our code:

 let job = client
        .job()
        .get_job(project_id, job_id, Some("europe-north1"))
        .await?;

[haxorcize]

@nixxholas @lquerel - any idea what's the difference between this method: https://cloud.google.com/bigquery/docs/reference/api-uploads - and the standard method of uploading to GCS and then starting a load job that copies from GCS?

Is the first method "directly to bigquery" and cheaper?