BigQuery sample: import data (streaming insert)

Author: remi Taylor

bigquery/spec/bigquery_sample_spec.rb

@@ -212,10 +212,8 @@ def capture &block
 
       expect(loaded_data).not_to be_empty
       expect(loaded_data.count).to eq 2
-      expect(loaded_data.first["name"]).to eq "Alice"
-      expect(loaded_data.first["value"]).to eq 5
-      expect(loaded_data.last["name"]).to eq "Bob"
-      expect(loaded_data.last["value"]).to eq 10
+      expect(loaded_data).to include({ "name" => "Alice", "value" => 5  })
+      expect(loaded_data).to include({ "name" => "Bob",   "value" => 10 })
     end
 
     example "import data from Cloud Storage" do
@@ -250,13 +248,34 @@ def capture &block
 
       expect(loaded_data).not_to be_empty
       expect(loaded_data.count).to eq 2
-      expect(loaded_data.first["name"]).to eq "Alice"
-      expect(loaded_data.first["value"]).to eq 5
-      expect(loaded_data.last["name"]).to eq "Bob"
-      expect(loaded_data.last["value"]).to eq 10
+      expect(loaded_data).to include({ "name" => "Alice", "value" => 5  })
+      expect(loaded_data).to include({ "name" => "Bob",   "value" => 10 })
     end
 
-    example "stream data import"
+    example "stream data import" do
+      expect(@table.data).to be_empty
+
+      row_data_to_insert = [
+        { name: "Alice", value: 5  },
+        { name: "Bob",   value: 10 }
+      ]
+
+      expect {
+        import_table_data project_id: @project_id,
+                          dataset_id: @dataset.dataset_id,
+                          table_id:   @table.table_id,
+                          row_data:   row_data_to_insert
+      }.to output(
+
+      ).to_stdout
+
+      loaded_data = @table.data
+
+      expect(loaded_data).not_to be_empty
+      expect(loaded_data.count).to eq 2
+      expect(loaded_data).to include({ "name" => "Alice", "value" => 5  })
+      expect(loaded_data).to include({ "name" => "Bob",   "value" => 10 })
+    end
   end
 
   describe "Exporting data" do

bigquery/tables.rb

@@ -87,6 +87,30 @@ def list_table_data project_id:, dataset_id:, table_id:
   # [END list_table_data]
 end
 
+def import_table_data project_id:, dataset_id:, table_id:, row_data:
+  # [START import_table_data]
+  # project_id = "Your Google Cloud project ID"
+  # dataset_id = "ID of the dataset containing table"
+  # table_id   = "ID of the table to import data into"
+  # row_data   = [{ column1: value, column2: value }, ...]
+
+  require "google/cloud"
+
+  gcloud   = Google::Cloud.new project_id
+  bigquery = gcloud.bigquery
+  dataset  = bigquery.dataset dataset_id
+  table    = dataset.table table_id
+
+  response = table.insert row_data
+
+  if response.success?
+    puts "Inserted rows successfully"
+  else
+    puts "Failed to insert #{response.error_rows.count} rows"
+  end
+  # [END import_table_data]
+end
+
 def import_table_data_from_file project_id:, dataset_id:, table_id:,
                                 local_file_path:
   # [START import_table_data_from_file]
@@ -117,7 +141,7 @@ def import_table_data_from_cloud_storage project_id:, dataset_id:, table_id:,
   # [START import_table_data_from_cloud_storage]
   # project_id   = "Your Google Cloud project ID"
   # dataset_id   = "ID of the dataset containing table"
-  # table_id     = "ID of the table to import file data into"
+  # table_id     = "ID of the table to import data into"
   # storage_path = "Storage path to file to import, eg. gs://bucket/file.csv"
 
   require "google/cloud"
@@ -204,6 +228,8 @@ def run_query_as_job project_id:, query_string:
 end
 
 if __FILE__ == $PROGRAM_NAME
+  require "json"
+
   project_id = ENV["GOOGLE_CLOUD_PROJECT"]
   command    = ARGV.shift
 
@@ -233,6 +259,11 @@ def run_query_as_job project_id:, query_string:
                                          dataset_id:   ARGV.shift,
                                          table_id:     ARGV.shift,
                                          storage_path: ARGV.shift
+  when "import_data"
+    import_table_data project_id: project_id,
+                      dataset_id: ARGV.shift,
+                      table_id:   ARGV.shift,
+                      row_data:   JSON.parse(ARGV.shift)
   when "export"
     export_table_data_to_cloud_storage project_id:   project_id,
                                        dataset_id:   ARGV.shift,
@@ -253,6 +284,7 @@ def run_query_as_job project_id:, query_string:
   list_data   <dataset_id> <table_id>  List data in table with the specified ID
   import_file <dataset_id> <table_id> <file_path>
   import_gcs  <dataset_id> <table_id> <cloud_storage_path>
+  import_data <dataset_id> <table_id> "[{ <json row data> }]"
   export      <dataset_id> <table_id> <cloud_storage_path>
   query       <query>
   query_job   <query>