連結至資料來源

Managed Service for Apache Spark 可與資料庫、資料倉儲和串流服務整合，處理多個系統的資料。本文將說明如何將 Managed Service for Apache Spark 叢集連結至 Cloud de Confiance by S3NS內外的資料來源。

事前準備

1. In the Cloud de Confiance console, on the project selector page, select or create a Cloud de Confiance 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

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

3. Verify that billing is enabled for your Cloud de Confiance project.

4. Enable the Managed Service for Apache Spark, BigQuery, and Cloud Storage APIs, if any are not already enabled.

   Roles required to enable APIs

   To enable APIs, you need the serviceusage.services.enable permission. If you created the project, then you likely already have this permission through the Owner role (roles/owner). Otherwise, you can get this permission through the Service Usage Admin role (roles/serviceusage.serviceUsageAdmin). Learn how to grant roles.

   Enable the APIs

5. 建立 Managed Service for Apache Spark 叢集。
6. 建立 Cloud Storage bucket。
7. 確認您具備存取資料來源的必要權限。

必要的角色

如要執行本頁面的範例，您必須具備特定 Identity and Access Management (IAM) 角色。視機構政策而定，系統可能已授予這些角色。如要查看角色授予情況，請參閱「是否需要授予角色？」一文。

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

使用者角色

如要取得建立 Managed Service for Apache Spark 叢集所需的權限，請要求管理員授予您下列 IAM 角色：

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

服務帳戶角色

為確保 Compute Engine 預設服務帳戶具備建立 Managed Service for Apache Spark 叢集的必要權限，請要求管理員在專案中，將 Dataproc 工作者 (roles/dataproc.worker) IAM 角色授予 Compute Engine 預設服務帳戶。

連結 Cloud de Confiance 服務

本節說明如何將 Managed Service for Apache Spark 連線至Cloud de Confiance 和外部資料來源。

連線至 Cloud Storage

Managed Service for Apache Spark 會使用 Cloud Storage 做為預設分散式檔案系統。如要從 Cloud Storage bucket 讀取及寫入資料，請使用 gs:// URI 前置字元。無須額外設定。

下列範例說明如何從 Cloud Storage bucket 讀取 CSV 檔案，然後將 Parquet 檔案寫入 bucket。

# Read a CSV file from a Cloud Storage bucket into a Spark DataFrame
df = spark.read.format("csv") \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .load("gs://BUCKET_NAME/path/to/data.csv")

df.show()

# Write a Spark DataFrame to a Cloud Storage bucket
df.write.format("parquet") \
    .mode("overwrite") \
    .save("gs://BUCKET_NAME/path/to/output.parquet")

連結至 BigQuery

在 Managed Service for Apache Spark 2.1 以上的映像檔版本中，系統會預先安裝 BigQuery 連接器。這個連接器可在 Managed Service for Apache Spark 與 BigQuery 之間，大規模傳輸資料。

以下範例說明如何將 BigQuery 資料表讀取到 Spark DataFrame，然後執行查詢。

# Set up the BigQuery connector.
spark.conf.set("spark.sql.sources.provider", "bigquery")
spark.conf.set("viewsEnabled", "true")
spark.conf.set("materializationDataset", "YOUR_BQ_DATASET")

# Read a BigQuery table into a Spark DataFrame,
df = spark.read.format("bigquery") \
    .option("table", "bigquery-public-data.samples.shakespeare") \
    .load()

df.createOrReplaceTempView("shakespeare")

# Perform a query on the data,
word_counts = spark.sql("SELECT word, SUM(word_count) as count FROM shakespeare GROUP BY word ORDER BY count DESC")
word_counts.show()

使用 JDBC 連線至外部資料庫

您可以使用 Spark 內建的 JDBC 資料來源，將 Managed Service for Apache Spark 連接至提供 Java Database Connectivity (JDBC) 驅動程式的資料庫，例如 PostgreSQL 或 MySQL。

提供 JDBC 驅動程式

如要透過 JDBC 連線至資料庫，請在 Spark 類路徑中提供驅動程式的 JAR 檔案。

gcloud dataproc clusters create my-jdbc-cluster \
    --region=REGION \
    --image-version=2.2-debian11 \
    --properties="spark:spark.jars=gs://YOUR_BUCKET_NAME/drivers/postgresql-42.7.3.jar"

從 PostgreSQL 資料庫讀取資料

這個範例使用每個工作的方法連線至 PostgreSQL 資料庫。

1. 下載 PostgreSQL JDBC 驅動程式 JAR，並上傳至 Cloud Storage bucket。

2. 提交 PySpark 工作，並使用 --jars 標記參照驅動程式的 Cloud Storage 路徑。

   gcloud dataproc jobs submit pyspark my_job.py \
       --cluster=MY_CLUSTER \
       --region=REGION \
       --jars=gs://YOUR_BUCKET_NAME/drivers/postgresql-42.7.3.jar
   

3. 在 job.py 檔案中使用下列程式碼，從資料庫讀取及寫入資料。

   from pyspark.sql import SparkSession
   
   def main():
       # WARNING: Do not hardcode credentials. Use a service like Secret Manager
       # to handle sensitive credentials.
       db_properties = {
           "user": "USERNAME",
           "password": "PASSWORD",
           "driver": "org.postgresql.Driver"
       }
   
       jdbc_url = "jdbc:postgresql://DB_HOST:DB_PORT/DATABASE"
   
       spark = SparkSession.builder.appName("Postgres JDBC Example").getOrCreate()
   
       # Read data from a PostgreSQL table
       df = spark.read.jdbc(url=jdbc_url, table="public.my_table", properties=db_properties)
   
       df.printSchema()
       df.show()
   
       # Write data to a new PostgreSQL table
       (df.write
           .jdbc(url=jdbc_url,
                 table="public.new_table",
                 mode="overwrite",
                 properties=db_properties))
   
   if __name__ == "__main__":
       main()
   

使用 ODBC 連線至外部資料庫

如要連線至沒有 JDBC 驅動程式的資料來源，請使用 Open Database Connectivity (ODBC) 驅動程式。建立叢集時，使用初始化動作在每個節點上安裝驅動程式及其依附元件。

這個範例說明如何連線至 Microsoft SQL Server 執行個體。

1. 建立初始化動作指令碼。

   1. 建立指令碼，安裝適用於 Debian 的 Microsoft SQL Server ODBC 驅動程式。

       #!/bin/bash
       # Initialization action for installing MS SQL ODBC driver on the cluster.
   
       set -e -x
   
       # Install dependencies for the driver.
       apt-get update
       apt-get install -y --no-install-recommends curl gnupg unixodbc-dev
   
       # Add Microsoft's official repository.
       curl https://packages.microsoft.com/keys/microsoft.asc | apt-key add -
       curl https://packages.microsoft.com/config/debian/11/prod.list > /etc/apt/sources.list.d/mssql-release.list
   
       # Install the driver.
       apt-get update
       ACCEPT_EULA=Y apt-get install -y msodbcsql18
   
       # Install the pyodbc library for Python.
       pip install pyodbc
       ```
   
   1.  Upload the script to a Cloud Storage bucket.
   
   ```shell
       gsutil cp install-odbc-mssql.sh gs://YOUR_BUCKET_NAME/init-actions/
       ```
   

2. 建立執行初始化動作的 Managed Service for Apache Spark 叢集。

   建立叢集時，請指向 Cloud Storage 中的初始化動作指令碼。

   gcloud dataproc clusters create my-odbc-cluster \
       --region=REGION \
       --image-version=2.2-debian11 \
       --initialization-actions=gs://BUCKET_NAME/init-actions/install-odbc-mssql.sh
   

3. 使用 PySpark 連線及查詢

   叢集執行後，即可使用 ODBC 驅動程式和 pyodbc 程式庫。下列程式碼會在 driver 節點上使用 pyodbc，將資料擷取到 Pandas DataFrame 中，然後轉換為 Spark DataFrame 以進行分散式處理。

   import pyodbc
   import pandas as pd
   from pyspark.sql import SparkSession
   
   def get_sql_data(connection_string, query):
       """
       Connects to the database using pyodbc, executes a query,
       and returns the result as a Pandas DataFrame.
       """
       cnxn = pyodbc.connect(connection_string)
       pdf = pd.read_sql(query, cnxn)
       cnxn.close()
       return pdf
   
   def main():
       spark = SparkSession.builder.appName("ODBC Example").getOrCreate()
   
       # WARNING: Do not hardcode credentials. Use Secret Manager.
       server = 'SERVER.database.windows.net'
       database = 'DATABASE'
       username = 'USERNAME'
       password = 'PASSWORD'
       driver = '{ODBC Driver 18 for SQL Server}' # The driver name must match the installed driver.
   
       # Create the connection string.
       connection_string = f'DRIVER={driver};SERVER=tcp:{server},1433;DATABASE={database};UID={username};PWD={password};Encrypt=yes;TrustServerCertificate=no;Connection Timeout=30;'
   
       sql_query = "SELECT * FROM Sales.Customer"
   
       # Fetch data as a Pandas DataFrame.
       pandas_df = get_sql_data(connection_string, sql_query)
   
       # Convert the Pandas DataFrame to a Spark DataFrame for distributed processing.
       spark_df = spark.createDataFrame(pandas_df)
   
       print("Successfully read data from SQL Server:")
       spark_df.printSchema()
       spark_df.show(5)
   
       # You can now perform distributed operations on the Spark DataFrame.
       print(f"Total number of customers: {spark_df.count()}")
   
   if __name__ == "__main__":
       main()
   

最佳做法

本節說明管理憑證和依附元件的最佳做法。

憑證管理

請勿在程式碼中以硬式編碼方式加入密碼或其他私密資訊。使用 Secret Manager 安全地儲存憑證，並從 Managed Service for Apache Spark 工作存取憑證。

依附元件管理

建立叢集時，可以使用初始化動作安裝驅動程式和程式庫。如果是複雜環境，建議您建立自訂的 Managed Service for Apache Spark 映像檔，並預先安裝所有依附元件。自訂映像檔可縮短叢集啟動時間，並確保一致性。

請先將公開來源的初始化動作指令碼複製到您自己的已設定版本 Cloud Storage bucket，再用於正式環境。請勿直接參照公開指令碼，因為這些指令碼可能會變更，且不會另行通知。

後續步驟

• 進一步瞭解 Secret Manager。
• 瞭解如何建立自訂映像檔。
• 進一步瞭解初始化動作。