Блог

Я пытаюсь вставить объект JSON в Google BigQuery, используя облачные функции, написанные на Java. Однако код выдает странную ошибку разрешения, и я хочу подтвердить, что облачная функция снабжена всеми разрешениями Bigquery для записи в таблицу. Идентификатор проекта, имя набора данных и имя таблицы также проверяются и являются правильными.

Я получаю исключение во время выполнения при выполнении приведенного ниже кода из функции Google Cloud. Пожалуйста, помогите. Ошибка:

2021-05-08 22:52:45.674 ISTTopicReaderGCPJFunctionesaj66v5ty43 OnError called: com.google.api.gax.rpc.PermissionDeniedException: io.grpc.StatusRuntimeException: PERMISSION_DENIED: Permission 'TABLES_UPDATE_DATA' denied on resource '<removed>' (or it may not exist). OnError called: com.google.api.gax.rpc.PermissionDeniedException: io.grpc.StatusRuntimeException: PERMISSION_DENIED: Permission 'TABLES_UPDATE_DATA' denied on resource '<removed>' (or it may not exist).

 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 d", 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 d",0));
        record.put("RPorganisationName", String.format("record d",1));
        record.put("RPdate", String.format("record d",2));
        record.put("RPtime", String.format("record d",3));
        record.put("RPmacid", String.format("record d",4));

        record.put("status", String.format("record d",5));
        record.put("mac", String.format("record d",6));
        record.put("date", String.format("record d",7));
        record.put("time", String.format("record d",8));
        record.put("count", String.format("record d",9));
        record.put("peakadc", String.format("record d",10));
        record.put("reset", String.format("record d",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 d", 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;
    }
  }
}
 

Что ж, это должно сработать для Вас.

Прежде всего, если вам нужны два облачных API/сервиса Google, например CF и BQ, для взаимодействия друг с другом, вам необходимо создать учетную запись службы, а затем предоставить этой учетной записи предопределенную роль, необходимую для взаимодействия с другими сервисами API, как в вашем случае (Администратор BigQuery — роли/bigquery.admin).

Решение:

1. Создайте учетную запись службы для CF.
2. Вы получите файл json, содержащий учетные данные и адрес электронной почты.
3. Дайте этот адрес электронной почты Администратору BigQuery — роли/bigquery.admin.
4. в вашем коде CF укажите все учетные данные, которые будут использоваться в вашем коде CF. Проверьте этот пример кода из Google cloud
5. Разверните Свой CF.

Этот процесс отлично работает со мной для Python.

Подробная информация :

Понимание учетных записей служб