fix: separate create_stream() in pyarrow sample (#946)

* fix: separate create_stream() in pyarrow sample

* remove append_rows()

* lint

* add blank row

Author: Linchin

bigquery_storage/pyarrow/append_rows_with_arrow.py

@@ -174,19 +174,6 @@ def generate_write_requests(pyarrow_table):
         yield request
 
 
-def append_rows(bqstorage_write_client, table):
-    append_rows_stream = create_stream(bqstorage_write_client, table)
-    pyarrow_table = generate_pyarrow_table()
-    futures = []
-
-    for request in generate_write_requests(pyarrow_table):
-        response_future = append_rows_stream.send(request)
-        futures.append(response_future)
-        response_future.result()
-
-    return futures
-
-
 def verify_result(client, table, futures):
     bq_table = client.get_table(table)
 
@@ -196,6 +183,7 @@ def verify_result(client, table, futures):
     # Verify table size.
     query = client.query(f"SELECT COUNT(1) FROM `{bq_table}`;")
     query_result = query.result().to_dataframe()
+
     # There might be extra rows due to retries.
     assert query_result.iloc[0, 0] >= TABLE_LENGTH
 
@@ -204,9 +192,28 @@ def verify_result(client, table, futures):
 
 
 def main(project_id, dataset):
+    # Initialize clients.
     write_client = bqstorage_write_client()
     bq_client = bigquery.Client()
-    table = make_table(project_id, dataset.dataset_id, bq_client)
 
-    futures = append_rows(write_client, table)
-    verify_result(bq_client, table, futures)
+    # Create BigQuery table.
+    bq_table = make_table(project_id, dataset.dataset_id, bq_client)
+
+    # Generate local PyArrow table.
+    pa_table = generate_pyarrow_table()
+
+    # Convert PyArrow table to Protobuf requests.
+    requests = generate_write_requests(pa_table)
+
+    # Create writing stream to the BigQuery table.
+    stream = create_stream(write_client, bq_table)
+
+    # Send requests.
+    futures = []
+    for request in requests:
+        future = stream.send(request)
+        futures.append(future)
+        future.result()  # Optional, will block until writing is complete.
+
+    # Verify results.
+    verify_result(bq_client, bq_table, futures)