使用 Gemini 開發 Spark 應用程式

本頁說明如何使用 Gemini CLI 開發 Apache Spark 應用程式，然後將應用程式提交至 Managed Service for Apache Spark 服務。

事前準備

登入 Google Cloud 帳戶。如果您是 Google Cloud新手，歡迎建立帳戶，親自評估產品在實際工作環境中的成效。新客戶還能獲得價值 $300 美元的免費抵免額，可用於執行、測試及部署工作負載。

In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

Roles required to select or create a project

Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

Go to project selector

Verify that you have the permissions required to complete this guide.

Verify that billing is enabled for your Google Cloud project.

Enable the Managed Service for Apache Spark, Cloud Storage, and BigQuery APIs.

Roles required to enable APIs

To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.

Enable the APIs

In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

Roles required to select or create a project

Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

Go to project selector

Verify that you have the permissions required to complete this guide.

Verify that billing is enabled for your Google Cloud project.

Enable the Managed Service for Apache Spark, Cloud Storage, and BigQuery APIs.

Roles required to enable APIs

To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.

Enable the APIs

必要的角色

如要執行本頁的範例，您必須具備特定 IAM 角色。視組織政策而定，系統可能已授予這些角色。如要檢查角色授予情形，請參閱「是否需要授予角色？」一節。

如要進一步瞭解如何授予角色，請參閱「管理專案、資料夾和機構的存取權」。

使用者角色

如要取得使用 Managed Service for Apache Spark 服務所需的權限，請要求系統管理員授予您下列 IAM 角色：

專案的 Dataproc 編輯者 (roles/dataproc.editor)
Compute Engine 預設服務帳戶的「服務帳戶使用者」 (roles/iam.serviceAccountUser)

服務帳戶角色

為確保 Compute Engine 預設服務帳戶具備提交 Managed Service for Apache Spark 工作及存取 Secret 的必要權限，請要求管理員在專案中將下列 IAM 角色授予 Compute Engine 預設服務帳戶：

提交工作： Dataproc 工作站 (roles/dataproc.worker)
存取密鑰： Secret Manager 密鑰存取者 (roles/secretmanager.secretAccessor)

Gemini 最佳做法

請遵循以下最佳做法：

啟動 Gemini 前，請在專案中設定程式碼開發環境，包括環境變數和 Python 虛擬環境。
將專案指南、程式碼樣式和架構儲存在 Gemini.md 檔案中。
請 Gemini 擬定計畫。
啟用查核點，即可復原至先前的狀態。
使用特定提示詞定義程式碼語言、架構、程式庫和結果。使用 @ 符號參照程式碼集檔案。
使用 ! 符號，從 Gemini CLI 執行殼層指令。

將 Hive 資料表遷移至 BigQuery

本節說明如何提示 Gemini 生成 PySpark 指令碼，將 Hive 資料表遷移至 BigQuery。

對 Gemini CLI 輸入下列提示詞：

Create a PySpark transform_hive_to_bigquery.py script to extract and transform a Hive
table, adding an insertion_time column using the add_insertion_time_column
function in @data_transformer.py. Save this table to BigQuery,
providing detailed instructions to run this script against a dataproc cluster.
Save a summary of this session to hive_to_BQReadme.md.

Gemini CLI 會產生類似下列內容的 transform_hive_to_bigquery.py 檔案：

  import argparse
  from pyspark.sql import SparkSession
  from data_transformer import add_insertion_time_column

  def transform_hive_to_bigquery(spark: SparkSession, hive_database: str, hive_table: str, bq_table: str, bq_temp_gcs_bucket: str):
      """
      Reads a Hive table, adds an insertion_time column, and writes to a BigQuery table.

      Args:
          spark: The SparkSession object.
          hive_database: The name of the source Hive database.
          hive_table: The name of the source Hive table.
          bq_table: The destination BigQuery table (e.g., 'dataset.table').
          bq_temp_gcs_bucket: The Cloud Storage bucket for temporary BigQuery
          connector data.
      """
      # Read data from Hive table.
      input_df = spark.table(f'{hive_database}.{hive_table}')

      # Add the insertion time column.
      transformed_df = add_insertion_time_column(input_df)

      # Write the transformed data to BigQuery.
      transformed_df.write \
          .format('bigquery') \
          .option('table', bq_table) \
          .option('temporaryGcsBucket', bq_temp_gcs_bucket) \
          .mode('append') \
          .save()

      print(f"Data successfully written to BigQuery table: {bq_table}")

  if __name__ == '__main__':
      parser = argparse.ArgumentParser(description='PySpark Hive to BigQuery transformation script')
      parser.add_argument('--hive_database', required=True, help='Source Hive database')
      parser.add_argument('--hive_table', required=True, help='Source Hive table')
      parser.add_argument('--bq_table', required=True, help='Destination BigQuery table (dataset.table)')
      parser.add_argument('--bq_temp_gcs_bucket', required=True, help='Cloud Storage bucket for temporary BigQuery connector data')
      args = parser.parse_args()

      # Create a SparkSession with Hive support.
      spark = SparkSession.builder \
          .appName('Hive to BigQuery Transformation') \
          .enableHiveSupport() \
          .getOrCreate()

      transform_hive_to_bigquery(spark, args.hive_database, args.hive_table, args.bq_table, args.bq_temp_gcs_bucket)

      spark.stop()

將 transform_hive_to_bigquery.py 檔案上傳至 Cloud Storage bucket (如需操作說明，請參閱「將物件上傳至 bucket」)。

執行下列指令，將工作提交至 Managed Service for Apache Spark 叢集：

gcloud dataproc jobs submit pyspark gs://BUCKET_NAME/transform_hive_to_bigquery.py \
    --cluster=CLUSTER_NAME
    --py-files=gs://BUCKET_NAME/data_transformer.py \
    --properties=spark.hadoop.hive.metastore.uris=METASTORE_URI \
    -- --hive_database=HIVE_DATABASE --hive_table=HIVE_TABLE --bq_table=BQ_DATASET.BQ_TABLE \
    --bq_temp_gcs_bucket=BUCKET_NAME

將資料表從 Postgres 遷移至 MySQL

本節說明如何提示 Gemini 生成 Java Spark 應用程式，將資料表從 Postgres 資料庫遷移至 MySQL 資料庫。

對 Gemini CLI 輸入下列提示詞：

Create a Spark `PostgresToMySql.java` job to migrate data from a table in a
Postgres database to a table in MySQL, both accessible via JDBC. The JDBC
URL strings are stored in Secret Manager. The URL string includes
the username and password. Read and write data in parallel based on
user-provided partitioning information. Write data in batches for efficiency.
Use the addInsertionTimeColumn to add a column to the data before writing it
to the MySQL destination table. Provide instructions to run this job on
Managed Service for Apache Spark in migrateJdbcToJdbc.md. Provide a
summary of the job in migration-README.md

Gemini 會檢查目錄，並在適當的套件中建立 Java 應用程式，以及用於建構應用程式的 pom.xml 檔案。PostgresToMySql.java 檔案會類似如下：

package com.customer.app;

import com.customer.util.DataframeUtils;
import com.google.cloud.secretmanager.v1.SecretManagerServiceClient;
import com.google.cloud.secretmanager.v1.SecretVersionName;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;

import java.io.IOException;

public class PostgresToMySql {

    public static void main(String[] args) {
        if (args.length != 6) {
            System.err.println("Usage: PostgresToMySql <postgres.table> <mysql.table> <postgres.secret.id> <mysql.secret.id> <partition.column> <batch.size> <project.id>");
            System.exit(1);
        }

        String postgresTable = args[0];
        String mysqlTable = args[1];
        String postgresSecretId = args[2];
        String mysqlSecretId = args[3];
        String partitionColumn = args[4];
        String batchSize = args[5];
        String projectId = args[6];

        SparkSession spark = SparkSession.builder()
                .appName("Postgres to MySQL JDBC Migration")
                .getOrCreate();

        String postgresUrl = getSecret(projectId, postgresSecretId);
        String mysqlUrl = getSecret(projectId, mysqlSecretId);

        // Read data from Postgres in parallel.
        Dataset<Row> df = spark.read()
                .format("jdbc")
                .option("url", postgresUrl)
                .option("dbtable", postgresTable)
                .option("partitionColumn", partitionColumn)
                .option("lowerBound", "2012-01-03")
                .option("upperBound", "2025-10-31") // These bounds should be configured based on data
                .option("numPartitions", "1")
                .option("driver", "org.postgresql.Driver")
                .load();

        // Add insertion time column.
        Dataset<Row> transformedDf = DataframeUtils.addInsertionTimeColumn(df);

        // Write data to MySQL in batches.
        transformedDf.write()
                .format("jdbc")
                .option("url", mysqlUrl)
                .option("dbtable", mysqlTable)
                .option("batchsize", Integer.parseInt(batchSize))
                .option("driver", "com.mysql.cj.jdbc.Driver")
                .mode(SaveMode.Overwrite)
                .save();

        spark.stop();
    }

    private static String getSecret(String secretId) {
        try (SecretManagerServiceClient client = SecretManagerServiceClient.create()) {
            SecretVersionName secretVersionName = SecretVersionName.of(projectId, secretId, "latest");
            return client.accessSecretVersion(secretVersionName).getPayload().getData().toStringUtf8();
        } catch (IOException e) {
            throw new RuntimeException("Failed to retrieve secret", e);
        }
    }
}

注意事項：

根據資料設定 lowerBound 和 upperBound。
如果產生的 pom.xml 無法正常運作，請使用這個 GitHub pom.xml。

將 postgres-to-mysql-migration-VERSION.jar 檔案上傳至 Cloud Storage bucket (如需操作說明，請參閱「將物件上傳至 bucket」)。

執行下列指令，將工作提交至 Managed Service for Apache Spark 叢集：

gcloud dataproc jobs submit spark
    --cluster=CLUSTER_NAME
    --class=com.customer.app.PostgresToMySql \
    --jars=BUCKET/postgres-to-mysql-migration-VERSION.jar \
    -- POSTGRES_TABLE MYSQL-TABLE \
    POSTGRES_SECRET MYSQL-SECRET COLUMN BATCH_SIZE