Java - 一次插入一行到谷歌大查询？

Posted 2023-03-24

【中文标题】Java - 一次插入一行到谷歌大查询？

【英文标题】：Java - Insert a single row at a time into google Big Query ?

【发布时间】：2018-05-08 16:49:11

【问题描述】：

我正在创建一个应用程序，在该应用程序中，每次用户点击一篇文章时，我都需要捕获文章数据和用户数据，以计算每篇文章的覆盖面，并能够对达到的数据进行分析。

我的应用程序在 App Engine 上。

当我检查插入 BQ 的文档时，其中大多数指向以作业或流形式的批量插入。

问题： 每次启动用户操作时一次插入大查询一行是否是一种好习惯？如果是这样，您能否指出一些 Java 代码来有效地做到这一点？

【参考方案1】：

加载作业和 DML 查询的数量有限制（每天 1,000 个），因此您需要为此类应用程序使用 streaming inserts。请注意，流式插入不同于从 Java 流中加载数据。

TableId tableId = TableId.of(datasetName, tableName);
// Values of the row to insert
Map<String, Object> rowContent = new HashMap<>();
rowContent.put("booleanField", true);
// Bytes are passed in base64
rowContent.put("bytesField", "Cg0NDg0="); // 0xA, 0xD, 0xD, 0xE, 0xD in base64
// Records are passed as a map
Map<String, Object> recordsContent = new HashMap<>();
recordsContent.put("stringField", "Hello, World!");
rowContent.put("recordField", recordsContent);
InsertAllResponse response =
    bigquery.insertAll(
        InsertAllRequest.newBuilder(tableId)
            .addRow("rowId", rowContent)
            // More rows can be added in the same RPC by invoking .addRow() on the builder
            .build());
if (response.hasErrors()) 
  // If any of the insertions failed, this lets you inspect the errors
  for (Entry<Long, List<BigQueryError>> entry : response.getInsertErrors().entrySet()) 
    // inspect row error
  

（来自https://cloud.google.com/bigquery/streaming-data-into-bigquery#bigquery-stream-data-java 的示例）

请特别注意，插入失败不会总是抛出异常。您还必须检查响应对象是否有错误。

每次启动用户操作时一次将一行插入到大查询中是否是一种好习惯？

是的，将事件流流式传输到 BigQuery 以进行分析是很常见的。如果您将多个事件缓冲到对 BigQuery 的同一个流式插入请求中，您可以获得更好的性能，但绝对支持一次一行。

【参考方案2】：

Google 示例的简化版本。

    Map<String, Object> row1Data = new HashMap<>();
    row1Data.put("booleanField", true);
    row1Data.put("stringField", "myString"); 

    Map<String, Object> row2Data = new HashMap<>();
    row2Data.put("booleanField", false);
    row2Data.put("stringField", "myOtherString"); 

    TableId tableId = TableId.of("myDatasetName", "myTableName");
    InsertAllResponse response =
            bigQuery.insertAll(
                    InsertAllRequest.newBuilder(tableId)
                            .addRow("row1Id", row1Data)
                            .addRow("row2Id", row2Data)
                            .build());

    if (response.hasErrors()) 
        // If any of the insertions failed, this lets you inspect the errors
        for (Map.Entry<Long, List<BigQueryError>> entry : response.getInsertErrors().entrySet()) 
            // inspect row error
        
    

【参考方案3】：

您可以使用 Cloud Logging API 一次写入一行。

https://cloud.google.com/logging/docs/reference/libraries

文档中的示例代码 公共类 QuickstartSample

/** 需要一个新的或现有的 Cloud 日志名称作为第一个参数。 */ public static void main(String... args) 抛出异常

// Instantiates a client
Logging logging = LoggingOptions.getDefaultInstance().getService();

// The name of the log to write to
String logName = args[0]; // "my-log";

// The data to write to the log
String text = "Hello, world!";
LogEntry entry =
    LogEntry.newBuilder(StringPayload.of(text))
        .setSeverity(Severity.ERROR)
        .setLogName(logName)
        .setResource(MonitoredResource.newBuilder("global").build())
        .build();

// Writes the log entry asynchronously
logging.write(Collections.singleton(entry));

System.out.printf("Logged: %s%n", text);
 

在这种情况下，您需要从数据流日志创建接收器。然后消息将被重定向到大查询表。

https://cloud.google.com/logging/docs/export/configure_export_v2