SpringCloud3.x集成BigQuery的代碼實(shí)現(xiàn)
1.原理
Google BigQuery 是一種高性能、可應(yīng)用于大數(shù)據(jù)分析的公主云數(shù)據(jù)庫服務(wù)。Spring Cloud 提供了完善的工具和核心功能,可以進(jìn)行泛動(dòng)分布應(yīng)用構(gòu)建。通過集成 Spring Cloud GCP,應(yīng)用可以便捷地使用 Google 云服務(wù),如 BigQuery。 運(yùn)行原理如下:
- Spring Cloud GCP 提供對 Google Cloud SDK 和 REST API 的封裝,通過自定義配置簡化了通信流程。
- 通過 BigQueryTemplate,應(yīng)用可以實(shí)現(xiàn)數(shù)據(jù)提交、查詢和分析。
- 使用 Spring 框架的常規(guī)構(gòu)件,如信息通道和任務(wù)調(diào)度,應(yīng)用可以進(jìn)行規(guī)?;瘮?shù)據(jù)處理。
2.應(yīng)用場景
- 大數(shù)據(jù)分析: 選擇 BigQuery 進(jìn)行大量數(shù)據(jù)的高性能分析,完善商業(yè)準(zhǔn)備和內(nèi)容提報(bào)。
- ETL 操作: 連接多種數(shù)據(jù)源,通過規(guī)則創(chuàng)建云數(shù)據(jù)分析的數(shù)據(jù)統(tǒng)一。
- BI 圖表: 使用 BigQuery 提供的高速查詢功能,支持 BI 平臺實(shí)現(xiàn)動(dòng)態(tài)繪圖和數(shù)據(jù)分析。
- 可視化報(bào)表: 在進(jìn)行精簡數(shù)據(jù)計(jì)算后,展示對外分析結(jié)果。
3.環(huán)境創(chuàng)建
到 BigQuery 信息中心,為自己創(chuàng)建一個(gè) BigQuery 數(shù)據(jù)集。在資源面板下,點(diǎn)擊您的項(xiàng)目 ID,然后點(diǎn)擊項(xiàng)目下的創(chuàng)建數(shù)據(jù)集。

4.代碼實(shí)現(xiàn)
1. 配置環(huán)境
1.1 創(chuàng)建項(xiàng)目
使用 Spring Initializr 創(chuàng)建 Spring Boot 項(xiàng)目,選擇以下依賴:
- Spring Web
- Spring Boot Actuator
- Spring Cloud GCP
1.2 加入依賴
在 pom.xml 中添加以下依賴:
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>spring-cloud-gcp</artifactId>
<groupId>com.et</groupId>
<version>1.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>spring-cloud-gcp-bigquery-sample</artifactId>
<properties>
<maven.compiler.source>17</maven.compiler.source>
<maven.compiler.target>17</maven.compiler.target>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>com.google.cloud</groupId>
<artifactId>spring-cloud-gcp-starter-bigquery</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-thymeleaf</artifactId>
</dependency>
<!-- Test-related dependencies. -->
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>
2. 配置文件
2.1 application.properties
spring.cloud.gcp.bigquery.dataset-name=test_dataset spring.cloud.gcp.bigquery.project-id=feisty-truth-447013-m7 spring.cloud.gcp.bigquery.credentials.location=file:/path-to-key/keyfile.json
舉例:請將 path-to-key/keyfile.json 替換為你的服務(wù)賬戶私鑰文件路徑。
2.2 IAM 訪問權(quán)限
確保對應(yīng)的服務(wù)賬戶具備以下角色:
BigQuery Data ViewerBigQuery Job User
使用以下指令加入 IAM 訪問權(quán)限:
gcloud projects add-iam-policy-binding PROJECT_ID \
--member="serviceAccount:YOUR_SERVICE_ACCOUNT_EMAIL" \
--role="roles/bigquery.jobUser"
替換 PROJECT_ID 和 YOUR_SERVICE_ACCOUNT_EMAIL 為你實(shí)際的項(xiàng)目 ID 和服務(wù)賬戶郵箱地址。
3. 實(shí)現(xiàn)邏輯
3.1 創(chuàng)建接口
將文件上傳至 BigQuery
/*
* Copyright 2017-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.et;
import com.et.BigQuerySampleConfiguration.BigQueryFileGateway;
import com.google.cloud.bigquery.*;
import com.google.cloud.spring.bigquery.core.BigQueryTemplate;
import com.google.cloud.spring.bigquery.core.WriteApiResponse;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Controller;
import org.springframework.ui.ModelMap;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.multipart.MultipartFile;
import org.springframework.web.servlet.ModelAndView;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.util.concurrent.CompletableFuture;
/** Provides REST endpoint allowing you to load data files to BigQuery using Spring Integration. */
@Controller
public class WebController {
private final BigQueryFileGateway bigQueryFileGateway;
private final BigQueryTemplate bigQueryTemplate;
private static final String DATASET_NAME = "datasetName";
@Value("${spring.cloud.gcp.bigquery.datasetName}")
private String datasetName;
public WebController(BigQueryFileGateway bigQueryFileGateway,
BigQueryTemplate bigQueryTemplate) {
this.bigQueryFileGateway = bigQueryFileGateway;
this.bigQueryTemplate = bigQueryTemplate;
}
@GetMapping("/")
public ModelAndView renderIndex(ModelMap map) {
map.put(DATASET_NAME, this.datasetName);
return new ModelAndView("index.html", map);
}
@GetMapping("/write-api-json-upload")
public ModelAndView renderUploadJson(ModelMap map) {
map.put(DATASET_NAME, this.datasetName);
return new ModelAndView("upload-json.html", map);
}
/**
* Handles a file upload using {@link BigQueryTemplate}.
*
* @param file the JSON file to upload to BigQuery
* @param tableName name of the table to load data into
* @return ModelAndView of the response the send back to users
* @throws IOException if the file is unable to be loaded.
*/
@PostMapping("/uploadJsonFile")
public ModelAndView handleJsonFileUpload(
@RequestParam("file") MultipartFile file,
@RequestParam("tableName") String tableName,
@RequestParam(name = "createTable", required = false) String createDefaultTable)
throws IOException {
CompletableFuture<WriteApiResponse> writeApiRes;
if (createDefaultTable != null
&& createDefaultTable.equals("createTable")) { // create the default table
writeApiRes =
this.bigQueryTemplate.writeJsonStream(
tableName, file.getInputStream(), getDefaultSchema());
} else { // we are expecting the table to be already existing
writeApiRes = this.bigQueryTemplate.writeJsonStream(tableName, file.getInputStream());
}
return getWriteApiResponse(writeApiRes, tableName);
}
private Schema getDefaultSchema() {
return Schema.of(
Field.of("CompanyName", StandardSQLTypeName.STRING),
Field.of("Description", StandardSQLTypeName.STRING),
Field.of("SerialNumber", StandardSQLTypeName.NUMERIC),
Field.of("Leave", StandardSQLTypeName.NUMERIC),
Field.of("EmpName", StandardSQLTypeName.STRING));
}
/**
* Handles JSON data upload using using {@link BigQueryTemplate}.
*
* @param jsonRows the String JSON data to upload to BigQuery
* @param tableName name of the table to load data into
* @return ModelAndView of the response the send back to users
*/
@PostMapping("/uploadJsonText")
public ModelAndView handleJsonTextUpload(
@RequestParam("jsonRows") String jsonRows,
@RequestParam("tableName") String tableName,
@RequestParam(name = "createTable", required = false) String createDefaultTable) {
CompletableFuture<WriteApiResponse> writeApiRes;
if (createDefaultTable != null
&& createDefaultTable.equals("createTable")) { // create the default table
writeApiRes =
this.bigQueryTemplate.writeJsonStream(
tableName, new ByteArrayInputStream(jsonRows.getBytes()), getDefaultSchema());
} else { // we are expecting the table to be already existing
writeApiRes =
this.bigQueryTemplate.writeJsonStream(
tableName, new ByteArrayInputStream(jsonRows.getBytes()));
}
return getWriteApiResponse(writeApiRes, tableName);
}
private ModelAndView getWriteApiResponse(
CompletableFuture<WriteApiResponse> writeApiFuture, String tableName) {
String message = null;
try {
WriteApiResponse apiResponse = writeApiFuture.get();
if (apiResponse.isSuccessful()) {
message = "Successfully loaded data to " + tableName;
} else if (apiResponse.getErrors() != null && !apiResponse.getErrors().isEmpty()) {
message =
String.format(
"Error occurred while loading the file, printing first error %s. Use WriteApiResponse.getErrors() to get the complete list of errors",
apiResponse.getErrors().get(0).getErrorMessage());
}
} catch (Exception e) {
e.printStackTrace();
message = "Error: " + e.getMessage();
}
return new ModelAndView("upload-json.html")
.addObject(DATASET_NAME, this.datasetName)
.addObject("message", message);
}
/**
* Handles a file upload using {@link BigQueryTemplate}.
*
* @param file the CSV file to upload to BigQuery
* @param tableName name of the table to load data into
* @return ModelAndView of the response to send back to users
* @throws IOException if the file is unable to be loaded.
*/
@PostMapping("/uploadFile")
public ModelAndView handleFileUpload(
@RequestParam("file") MultipartFile file, @RequestParam("tableName") String tableName)
throws IOException {
CompletableFuture<Job> loadJob =
this.bigQueryTemplate.writeDataToTable(
tableName, file.getInputStream(), FormatOptions.csv());
return getResponse(loadJob, tableName);
}
/**
* Handles CSV data upload using Spring Integration {@link BigQueryFileGateway}.
*
* @param csvData the String CSV data to upload to BigQuery
* @param tableName name of the table to load data into
* @return ModelAndView of the response the send back to users
*/
@PostMapping("/uploadCsvText")
public ModelAndView handleCsvTextUpload(
@RequestParam("csvText") String csvData, @RequestParam("tableName") String tableName) {
CompletableFuture<Job> loadJob =
this.bigQueryFileGateway.writeToBigQueryTable(csvData.getBytes(), tableName);
return getResponse(loadJob, tableName);
}
private ModelAndView getResponse(CompletableFuture<Job> loadJob, String tableName) {
String message;
try {
Job job = loadJob.get();
message = "Successfully loaded data file to " + tableName;
} catch (Exception e) {
e.printStackTrace();
message = "Error: " + e.getMessage();
}
return new ModelAndView("index")
.addObject(DATASET_NAME, this.datasetName)
.addObject("message", message);
}
}
以上只是一些關(guān)鍵代碼。
5.測試
在google cloud shell里面運(yùn)行代碼 運(yùn)行 $ mvn spring-boot:run 命令。

單擊 Cloud Shell 中的 Web Preview 按鈕以在端口 8080 上預(yù)覽應(yīng)用,并嘗試將一些數(shù)據(jù)加載到數(shù)據(jù)集下的 BigQuery 表中。該應(yīng)用程序接受 CSV 文件上傳或輸入到文本區(qū)域的 CSV 數(shù)據(jù)。如果 BigQuery 數(shù)據(jù)集下尚不存在該表,則會(huì)為您創(chuàng)建該表。

查看導(dǎo)入結(jié)果

以上就是SpringCloud3.x集成BigQuery的代碼實(shí)現(xiàn)的詳細(xì)內(nèi)容,更多關(guān)于SpringCloud3.x集成BigQuery的資料請關(guān)注腳本之家其它相關(guān)文章!
相關(guān)文章
springboot + jpa實(shí)現(xiàn)刪除數(shù)據(jù)的操作代碼
這篇文章主要介紹了springboot + jpa實(shí)現(xiàn)刪除數(shù)據(jù)的操作代碼,本文通過實(shí)例代碼給大家介紹的非常詳細(xì),對大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值,需要的朋友參考下吧2024-05-05
SpringBoot通過Filter實(shí)現(xiàn)整個(gè)項(xiàng)目接口的SQL注入攔截詳解
這篇文章主要介紹了SpringBoot通過Filter實(shí)現(xiàn)整個(gè)項(xiàng)目接口的SQL注入攔截詳解,SQL注入是比較常見的網(wǎng)絡(luò)攻擊方式之一,在客戶端在向服務(wù)器發(fā)送請求的時(shí)候,sql命令通過表單提交或者url字符串拼接傳遞到后臺持久層,最終達(dá)到欺騙服務(wù)器執(zhí)行惡意的SQL命令,需要的朋友可以參考下2023-12-12
Java版仿QQ驗(yàn)證碼風(fēng)格圖片驗(yàn)證碼
這篇文章主要為大家分享了java圖片驗(yàn)證碼實(shí)例代碼,感興趣的小伙伴們可以參考一下2016-04-04
springboot整合xxl-job實(shí)現(xiàn)分布式定時(shí)任務(wù)的過程
XXL-JOB是一個(gè)分布式任務(wù)調(diào)度平臺,其核心設(shè)計(jì)目標(biāo)是開發(fā)迅速、學(xué)習(xí)簡單、輕量級、易擴(kuò)展,這篇文章主要介紹了springboot整合xxl-job分布式定時(shí)任務(wù),需要的朋友可以參考下2022-08-08
MyBatis 參數(shù)類型為String時(shí)常見問題及解決方法
這篇文章主要介紹了MyBatis 參數(shù)類型為String時(shí)常見問題及解決方法,非常不錯(cuò),具有參考借鑒價(jià)值,需要的朋友可以參考下2017-03-03
Spring?Data?JPA?注解Entity關(guān)聯(lián)關(guān)系使用詳解
這篇文章主要為大家介紹了Spring?Data?JPA?注解Entity關(guān)聯(lián)關(guān)系使用詳解,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2022-09-09

