问PERMISSION_DENIED:权限TABLES_UPDATE_DATA

EN

package functions;

import com.google.api.core.ApiFuture;
import com.google.cloud.bigquery.BigQuery;
import com.google.cloud.bigquery.BigQueryOptions;
import com.google.cloud.bigquery.Schema;
import com.google.cloud.bigquery.Table;
import com.google.cloud.bigquery.storage.v1beta2.AppendRowsResponse;
import com.google.cloud.bigquery.storage.v1beta2.BigQueryWriteClient;
import com.google.cloud.bigquery.storage.v1beta2.CreateWriteStreamRequest;
import com.google.cloud.bigquery.storage.v1beta2.JsonStreamWriter;
import com.google.cloud.bigquery.storage.v1beta2.TableName;
import com.google.cloud.bigquery.storage.v1beta2.WriteStream;
import com.google.protobuf.Descriptors.DescriptorValidationException;
import java.io.IOException;
import java.util.concurrent.ExecutionException;
import org.json.JSONArray;
import org.json.JSONObject;

//import org.json.simple.JSONArray;
//import org.json.simple.JSONObject;
//import org.json.simple.parser.JSONParser;

import java.util.*;

public class WriteCommittedStream {
  // TODO(developer): Replace these variables before running the sample.
  public static boolean writeCommittedStreamToBQ(String projectId, String datasetName, String tableName,
      HashSet<JSONArray> streamHashSet) throws DescriptorValidationException, InterruptedException, IOException {

    try (BigQueryWriteClient client = BigQueryWriteClient.create()) {

      Iterator<JSONArray> value = streamHashSet.iterator();
      while (value.hasNext()) {
        // Create a JSON object that is compatible with the table schema.
        // JSONObject record = new JSONObject();

        // record.put("col1", String.format("record %03d", value.next()));
        // JSONArray jsonArr = new JSONArray();
        // value.next().
        // jsonArr.get(value.next());
        System.out.println(value.next().get(0).toString());
      }
      // Array -> {data}
      // Array -> [{data}]
      System.out.println("projectId:" + projectId);
      System.out.println("datasetName:" + datasetName);
      System.out.println("tableName:" + tableName);

      BigQuery bigquery = BigQueryOptions.getDefaultInstance().getService();
      Table table = bigquery.getTable(datasetName, tableName);
      TableName parentTable = TableName.of(projectId, datasetName, tableName);
      Schema schema = table.getDefinition().getSchema();

      System.out.println("Schema:" + schema.toString());
      System.out.println("Table:" + parentTable.toString());

      try (JsonStreamWriter writer = JsonStreamWriter.newBuilder(parentTable.toString(), schema).createDefaultStream()
          .build()) {
       
        // Append 10 JSON objects to the stream.
        Iterator<JSONArray> value2 = streamHashSet.iterator();

        /////////////////////
/*
        JSONArray jsonArr = value2.next();
        System.out.println("Inside the loop:"+jsonArr);
        ApiFuture<AppendRowsResponse> future = writer.append(jsonArr);
        AppendRowsResponse response = future.get();
        System.out.println("Appended records successfully." + response.toString());
*/      
        JSONObject record = new JSONObject();
        record.put("RPdeviceName", String.format("record %03d",0));
        record.put("RPorganisationName", String.format("record %03d",1));
        record.put("RPdate", String.format("record %03d",2));
        record.put("RPtime", String.format("record %03d",3));
        record.put("RPmacid", String.format("record %03d",4));

        record.put("status", String.format("record %03d",5));
        record.put("mac", String.format("record %03d",6));
        record.put("date", String.format("record %03d",7));
        record.put("time", String.format("record %03d",8));
        record.put("count", String.format("record %03d",9));
        record.put("peakadc", String.format("record %03d",10));
        record.put("reset", String.format("record %03d",11));
        

        JSONArray jsonArr = new JSONArray();
        jsonArr.put(record);
        System.out.println("Initially." + jsonArr.toString());

        ApiFuture<AppendRowsResponse> future = writer.append(jsonArr);
        AppendRowsResponse response = future.get();
        System.out.println("Appended records successfully." + response.toString());
        //////////////////////
            /*
        while (value2.hasNext()) {
          // Create a JSON object that is compatible with the table schema.
          // JSONObject record = new JSONObject();

          // record.put("col1", String.format("record %03d", value.next()));
          JSONArray jsonArr = value2.next();
          // jsonArr.put(value2.next());
          // System.out.println(jsonArr.get(0).toString());

          System.out.println("Inside the loop:"+jsonArr);
          // jsonArr =

          // To detect duplicate records, pass the index as the record offset.
          // To disable deduplication, omit the offset or use WriteStream.Type.DEFAULT.
          ApiFuture<AppendRowsResponse> future = writer.append(jsonArr);
          
          AppendRowsResponse response = future.get();
          System.out.println("Appended records successfully." + response.toString());
        }*/
      }

      return true;
    } catch (ExecutionException e) {
      // If the wrapped exception is a StatusRuntimeException, check the state of the
      // operation.
      // If the state is INTERNAL, CANCELLED, or ABORTED, you can retry. For more
      // information, see:
      // https://grpc.github.io/grpc-java/javadoc/io/grpc/StatusRuntimeException.html
      System.out.println("Failed to append records. \n" );
      e.printStackTrace();
      return false;
    }
  }
}