Dịch vụ được quản lý cho Apache Spark

1. Tổng quan – Serverless cho Apache Spark

Dịch vụ được quản lý cho Apache Spark là một dịch vụ được quản lý toàn diện và có khả năng mở rộng cao để chạy Apache Spark, Apache Flink, Presto và nhiều công cụ cũng như khung nguồn mở khác. Sử dụng Apache Spark được quản lý để hiện đại hoá hồ dữ liệu, ETL / ELT và bảo mật khoa học dữ liệu ở quy mô toàn cầu. Apache Spark được quản lý cũng được tích hợp đầy đủ với một số dịch vụ của Google Cloud, bao gồm BigQuery, Cloud Storage, Gemini Enterprise Agent Engine và Knowledge Catalog.

Apache Spark được quản lý có 2 phiên bản:

• Apache Spark không máy chủ được quản lý cho phép bạn chạy các công việc PySpark mà không cần định cấu hình cơ sở hạ tầng và tính năng tự động mở rộng quy mô. Apache Spark được quản lý hỗ trợ các phiên / sổ tay và khối lượng công việc hàng loạt PySpark.
• Cụm Apache Spark được quản lý cho phép bạn quản lý một cụm Hadoop YARN cho các tải công việc Spark dựa trên YARN, ngoài các công cụ nguồn mở như Flink và Presto. Bạn có thể điều chỉnh các cụm dựa trên đám mây theo ý muốn, bao gồm cả tính năng tự động mở rộng quy mô theo chiều dọc hoặc chiều ngang.

Trong lớp học lập trình này, bạn sẽ tìm hiểu một số cách để sử dụng Dataproc Serverless.

Ban đầu, Apache Spark được xây dựng để chạy trên các cụm Hadoop và sử dụng YARN làm trình quản lý tài nguyên. Việc duy trì các cụm Hadoop đòi hỏi một bộ kiến thức chuyên môn cụ thể và đảm bảo nhiều nút khác nhau trên các cụm được định cấu hình đúng cách. Điều này bổ sung cho một bộ nút riêng biệt mà Spark cũng yêu cầu người dùng thiết lập. Điều này dẫn đến nhiều trường hợp mà nhà phát triển dành nhiều thời gian hơn để định cấu hình cơ sở hạ tầng của họ thay vì làm việc trên chính mã Spark.

Dataproc Serverless giúp bạn không cần phải định cấu hình cụm Hadoop hoặc Spark theo cách thủ công. Dataproc Serverless không chạy trên Hadoop và sử dụng tính năng Phân bổ tài nguyên linh hoạt của riêng mình để xác định các yêu cầu về tài nguyên, bao gồm cả phương thức tự động cấp tài nguyên bổ sung. Một nhóm nhỏ các thuộc tính Spark vẫn có thể tuỳ chỉnh bằng Dataproc Serverless, tuy nhiên, trong hầu hết các trường hợp, bạn sẽ không cần điều chỉnh các thuộc tính này.

2. Thiết lập

Bạn sẽ bắt đầu bằng cách định cấu hình môi trường và tài nguyên được dùng trong lớp học lập trình này.

Tạo một dự án trên Google Cloud. Bạn có thể sử dụng một tài khoản hiện có.

Mở Cloud Shell bằng cách nhấp vào biểu tượng này trên thanh công cụ Cloud Console.

Cloud Shell cung cấp một môi trường Shell sẵn sàng sử dụng mà bạn có thể dùng cho lớp học lập trình này.

Cloud Shell sẽ đặt tên dự án của bạn theo mặc định. Kiểm tra kỹ bằng cách chạy echo $GOOGLE_CLOUD_PROJECT. Nếu bạn không thấy mã dự án trong đầu ra, hãy đặt mã dự án.

export GOOGLE_CLOUD_PROJECT=<your-project-id>

Đặt khu vực Compute Engine cho các tài nguyên của bạn, chẳng hạn như us-central1 hoặc europe-west2.

export REGION=<your-region>

Bật API

Lớp học lập trình này sử dụng các API sau:

• BigQuery
• Dataproc

Bật các API cần thiết. Quá trình này sẽ mất khoảng một phút và thông báo thành công sẽ xuất hiện khi hoàn tất.

gcloud services enable bigquery.googleapis.com
gcloud services enable dataproc.googleapis.com

Định cấu hình quyền truy cập mạng

Dataproc Serverless yêu cầu bạn bật Quyền truy cập riêng tư của Google ở khu vực mà bạn sẽ chạy các công việc Spark vì trình điều khiển và trình thực thi Spark chỉ có IP riêng tư. Chạy lệnh sau để bật tính năng này trong mạng con default.

gcloud compute networks subnets update default \
  --region=${REGION} \
  --enable-private-ip-google-access

Bạn có thể xác minh rằng Google Private Access đã được bật thông qua lệnh sau đây. Lệnh này sẽ xuất ra True hoặc False.

gcloud compute networks subnets describe default \
  --region=${REGION} \
  --format="get(privateIpGoogleAccess)"

Tạo bộ chứa lưu trữ

Tạo một bộ chứa lưu trữ sẽ được dùng để lưu trữ các thành phần được tạo trong lớp học lập trình này.

Chọn tên cho nhóm của bạn. Tên bộ chứa phải là duy nhất trên toàn cầu đối với tất cả người dùng.

export BUCKET=<your-bucket-name>

Tạo vùng chứa ở khu vực mà bạn dự định chạy các công việc Spark.

gsutil mb -l ${REGION} gs://${BUCKET}

Bạn có thể thấy bộ chứa của mình trong bảng điều khiển Cloud Storage. Bạn cũng có thể chạy gsutil ls để xem nhóm của mình.

Tạo một Máy chủ nhật ký liên tục

Giao diện người dùng Spark cung cấp một bộ công cụ gỡ lỗi và thông tin chi tiết về các công việc của Spark. Để xem Giao diện người dùng Spark cho các công việc đã hoàn thành của Dataproc Serverless, bạn phải tạo một cụm Dataproc một nút để sử dụng làm máy chủ nhật ký liên tục.

Đặt tên cho máy chủ nhật ký liên tục.

PHS_CLUSTER_NAME=my-phs

Chạy lệnh sau.

gcloud dataproc clusters create ${PHS_CLUSTER_NAME} \
    --region=${REGION} \
    --single-node \
    --enable-component-gateway \
    --properties=spark:spark.history.fs.logDirectory=gs://${BUCKET}/phs/*/spark-job-history

Giao diện người dùng Spark và máy chủ nhật ký liên tục sẽ được khám phá chi tiết hơn ở phần sau của lớp học lập trình.

3. Chạy các công việc Spark không máy chủ bằng Dataproc Batches

Trong mẫu này, bạn sẽ làm việc với một tập hợp dữ liệu từ tập dữ liệu công khai về Chuyến đi bằng xe đạp Citi ở Thành phố New York (NYC). NYC Citi Bikes là một hệ thống xe đạp công cộng có tính phí ở Thành phố New York. Bạn sẽ thực hiện một số phép biến đổi đơn giản và in 10 mã nhận dạng trạm Citi Bike phổ biến nhất. Đáng chú ý là mẫu này cũng sử dụng spark-bigquery-connector nguồn mở để đọc và ghi dữ liệu liền mạch giữa Spark và BigQuery.

Sao chép kho lưu trữ Github và cd sau đây vào thư mục chứa tệp citibike.py.

git clone https://github.com/GoogleCloudPlatform/devrel-demos.git
cd devrel-demos/data-analytics/next-2022-workshop/dataproc-serverless

citibike.py

import sys

from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import BooleanType

if len(sys.argv) == 1:
    print("Please provide a GCS bucket name.")

bucket = sys.argv[1]
table = "bigquery-public-data:new_york_citibike.citibike_trips"

spark = SparkSession.builder \
          .appName("pyspark-example") \
          .config("spark.jars","gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.26.0.jar") \
          .getOrCreate()

df = spark.read.format("bigquery").load(table)

top_ten = df.filter(col("start_station_id") \
            .isNotNull()) \
            .groupBy("start_station_id") \
            .count() \
            .orderBy("count", ascending=False) \
            .limit(10) \
            .cache()

top_ten.show()

top_ten.write.option("header", True).csv(f"gs://{bucket}/citibikes_top_ten_start_station_ids")

Gửi công việc đến Serverless Spark bằng Cloud SDK (theo mặc định có trong Cloud Shell). Chạy lệnh sau trong shell để sử dụng Cloud SDK và Dataproc Batches API nhằm gửi các công việc Serverless Spark.

gcloud dataproc batches submit pyspark citibike.py \
  --batch=citibike-job \
  --region=${REGION} \
  --deps-bucket=gs://${BUCKET} \
  --jars=gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.26.0.jar \
--history-server-cluster=projects/${GOOGLE_CLOUD_PROJECT}/regions/${REGION}/clusters/${PHS_CLUSTER_NAME} \
  -- ${BUCKET}

Để phân tích chi tiết:

• gcloud dataproc batches submit tham chiếu đến Dataproc Batches API.
• pyspark cho biết bạn đang gửi một công việc PySpark.
• --batch là tên của công việc. Nếu bạn không cung cấp, hệ thống sẽ sử dụng một mã UUID được tạo ngẫu nhiên.
• --region=${REGION} là khu vực địa lý nơi công việc sẽ được xử lý.
• --deps-bucket=${BUCKET} là nơi tệp Python cục bộ của bạn được tải lên trước khi chạy trong môi trường Serverless.
• --jars=gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.26.0.jar bao gồm tệp jar cho spark-bigquery-connector trong môi trường thời gian chạy Spark.
• --history-server-cluster=projects/${GOOGLE_CLOUD_PROJECT}/regions/${REGION}/clusters/${PHS_CLUSTER} là tên đủ điều kiện của máy chủ nhật ký liên tục. Đây là nơi lưu trữ dữ liệu sự kiện Spark (tách biệt với đầu ra của bảng điều khiển) và có thể xem được từ giao diện người dùng Spark.
• -- ở cuối cho biết mọi thứ sau đây sẽ là đối số thời gian chạy cho chương trình. Trong trường hợp này, bạn đang gửi tên của nhóm theo yêu cầu của công việc.

Bạn sẽ thấy kết quả sau đây khi gửi lô.

Batch [citibike-job] submitted.

Sau vài phút, bạn sẽ thấy đầu ra sau đây cùng với siêu dữ liệu của công việc.

+----------------+------+
|start_station_id| count|
+----------------+------+
|             519|551078|
|             497|423334|
|             435|403795|
|             426|384116|
|             293|372255|
|             402|367194|
|             285|344546|
|             490|330378|
|             151|318700|
|             477|311403|
+----------------+------+

Batch [citibike-job] finished.

Trong phần tiếp theo, bạn sẽ tìm hiểu cách xác định vị trí của nhật ký cho công việc này.

Các tính năng khác

Với Spark Serverless, bạn có thêm các lựa chọn để chạy các công việc của mình.

• Bạn có thể tạo một hình ảnh docker tuỳ chỉnh mà công việc của bạn sẽ chạy trên đó. Đây là một cách tuyệt vời để thêm các phần phụ thuộc khác, bao gồm cả thư viện Python và R.
• Bạn có thể kết nối một phiên bản Dataproc Metastore với công việc của mình để truy cập vào siêu dữ liệu Hive.
• Để có thêm quyền kiểm soát, Dataproc Serverless hỗ trợ việc định cấu hình một nhóm nhỏ thuộc tính Spark.

4. Chỉ số và khả năng quan sát của Dataproc

Dataproc Batches Console liệt kê tất cả các công việc Dataproc Serverless của bạn. Trong bảng điều khiển, bạn sẽ thấy Mã lô, Vị trí, Trạng thái, Thời gian tạo, Thời gian đã trôi qua và Loại của từng công việc. Nhấp vào Mã lô của công việc để xem thêm thông tin về công việc đó.

Trên trang này, bạn sẽ thấy thông tin như Giám sát cho biết số lượng Trình thực thi Spark theo lô mà công việc của bạn đã sử dụng theo thời gian (cho biết mức độ tự động mở rộng quy mô).

Trên thẻ Chi tiết, bạn sẽ thấy thêm siêu dữ liệu về công việc, bao gồm mọi đối số và tham số đã được gửi cùng với công việc.

Bạn cũng có thể truy cập vào tất cả nhật ký trên trang này. Khi các công việc Dataproc Serverless được chạy, 3 nhóm nhật ký khác nhau sẽ được tạo:

• Cấp dịch vụ
• Đầu ra trên bảng điều khiển
• Ghi nhật ký sự kiện Spark

Cấp dịch vụ, bao gồm nhật ký do dịch vụ Dataproc không dùng máy chủ tạo. Ví dụ: Dataproc Serverless yêu cầu thêm CPU để tự động mở rộng quy mô. Bạn có thể xem các nhật ký này bằng cách nhấp vào Xem nhật ký. Thao tác này sẽ mở Cloud Logging.

Bạn có thể xem đầu ra của bảng điều khiển trong phần Đầu ra.Đây là đầu ra do lệnh tạo ra, bao gồm cả siêu dữ liệu mà Spark in khi bắt đầu một lệnh hoặc bất kỳ câu lệnh in nào được đưa vào lệnh.

Bạn có thể truy cập vào Nhật ký sự kiện Spark thông qua giao diện người dùng Spark. Vì bạn đã cung cấp cho công việc Spark của mình một máy chủ nhật ký liên tục, nên bạn có thể truy cập vào giao diện người dùng Spark bằng cách nhấp vào Xem máy chủ nhật ký Spark. Máy chủ này chứa thông tin về các công việc Spark mà bạn đã chạy trước đó. Bạn có thể tìm hiểu thêm về giao diện người dùng Spark trong tài liệu chính thức về Spark.

5. Mẫu Dataproc: BQ -> GCS

Mẫu Dataproc là các công cụ nguồn mở giúp đơn giản hoá hơn nữa các tác vụ xử lý dữ liệu trên đám mây. Các mẫu này đóng vai trò là một trình bao bọc cho Dataproc Serverless và bao gồm các mẫu cho nhiều nhiệm vụ nhập và xuất dữ liệu, bao gồm:

• BigQuerytoGCS và GCStoBigQuery
• GCStoBigTable
• GCStoJDBC và JDBCtoGCS
• HivetoBigQuery
• MongotoGCS và GCStoMongo

Bạn có thể xem danh sách đầy đủ trong README.

Trong phần này, bạn sẽ sử dụng Dataproc Templates để xuất dữ liệu từ BigQuery sang GCS.

Sao chép kho lưu trữ

Sao chép kho lưu trữ và chuyển vào thư mục python.

git clone https://github.com/GoogleCloudPlatform/dataproc-templates.git
cd dataproc-templates/python

Định cấu hình môi trường

Bây giờ, bạn sẽ thiết lập các biến môi trường. Mẫu Dataproc sử dụng biến môi trường GCP_PROJECT cho mã dự án của bạn, vì vậy hãy đặt biến này bằng GOOGLE_CLOUD_PROJECT.

export GCP_PROJECT=${GOOGLE_CLOUD_PROJECT}

Bạn nên đặt khu vực trong môi trường từ trước đó. Nếu chưa, hãy thiết lập tại đây.

export REGION=<region>

Dataproc Templates sử dụng spark-bigquery-conector để xử lý các công việc BigQuery và yêu cầu phải có URI trong một biến môi trường JARS. Đặt biến JARS.

export JARS="gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.26.0.jar"

Định cấu hình các thông số mẫu

Đặt tên cho một vùng lưu trữ tạm thời để dịch vụ sử dụng.

export GCS_STAGING_LOCATION=gs://${BUCKET}

Tiếp theo, bạn sẽ thiết lập một số biến dành riêng cho công việc. Đối với bảng đầu vào, bạn sẽ tham chiếu lại tập dữ liệu BigQuery NYC Citibike.

BIGQUERY_GCS_INPUT_TABLE=bigquery-public-data.new_york_citibike.citibike_trips

Bạn có thể chọn csv, parquet, avro hoặc json. Đối với lớp học lập trình này, hãy chọn CSV. Trong phần tiếp theo, bạn sẽ tìm hiểu cách sử dụng Dataproc Templates để chuyển đổi các loại tệp.

BIGQUERY_GCS_OUTPUT_FORMAT=csv

Đặt chế độ đầu ra thành overwrite. Bạn có thể chọn overwrite, append, ignore hoặc errorifexists.

BIGQUERY_GCS_OUTPUT_MODE=overwrite

Đặt vị trí đầu ra GCS thành một đường dẫn trong bộ chứa của bạn.

BIGQUERY_GCS_OUTPUT_LOCATION=gs://${BUCKET}/BQtoGCS

Chạy mẫu

Chạy mẫu BIGQUERYTOGCS bằng cách chỉ định mẫu đó bên dưới và cung cấp các tham số đầu vào mà bạn đã đặt.

./bin/start.sh \
-- --template=BIGQUERYTOGCS \
        --bigquery.gcs.input.table=${BIGQUERY_GCS_INPUT_TABLE} \
        --bigquery.gcs.output.format=${BIGQUERY_GCS_OUTPUT_FORMAT} \
        --bigquery.gcs.output.mode=${BIGQUERY_GCS_OUTPUT_MODE} \
        --bigquery.gcs.output.location=${BIGQUERY_GCS_OUTPUT_LOCATION}

Đầu ra sẽ khá ồn, nhưng sau khoảng một phút, bạn sẽ thấy nội dung sau.

Batch [5766411d6c78444cb5e80f305308d8f8] submitted.
...
Batch [5766411d6c78444cb5e80f305308d8f8] finished.

Bạn có thể xác minh rằng các tệp được tạo bằng cách chạy lệnh sau.

gsutil ls ${BIGQUERY_GCS_OUTPUT_LOCATION}

Theo mặc định, Spark sẽ ghi vào nhiều tệp, tuỳ thuộc vào lượng dữ liệu. Trong trường hợp này, bạn sẽ thấy khoảng 30 tệp được tạo. Tên tệp đầu ra của Spark được định dạng bằng part, theo sau là một số có 5 chữ số (cho biết số phần) và một chuỗi băm. Đối với lượng dữ liệu lớn, Spark thường sẽ ghi vào một số tệp. Ví dụ về tên tệp là part-00000-cbf69737-867d-41cc-8a33-6521a725f7a0-c000.csv.

6. Mẫu Dataproc: CSV sang Parquet

Giờ đây, bạn sẽ sử dụng Dataproc Templates để chuyển đổi dữ liệu trong GCS từ một loại tệp sang một loại tệp khác bằng cách sử dụng GCSTOGCS. Mẫu này sử dụng SparkSQL và cung cấp lựa chọn gửi truy vấn SparkSQL để được xử lý trong quá trình chuyển đổi nhằm xử lý thêm.

Xác nhận các biến môi trường

Xác nhận rằng GCP_PROJECT, REGION và GCS_STAGING_BUCKET được đặt từ phần trước.

echo ${GCP_PROJECT}
echo ${REGION}
echo ${GCS_STAGING_LOCATION}

Đặt thông số mẫu

Bây giờ, bạn sẽ đặt các thông số cấu hình cho GCStoGCS. Bắt đầu bằng vị trí của các tệp đầu vào. Xin lưu ý rằng đây là một thư mục chứ không phải một tệp cụ thể vì tất cả các tệp trong thư mục sẽ được xử lý. Đặt giá trị này thành BIGQUERY_GCS_OUTPUT_LOCATION.

GCS_TO_GCS_INPUT_LOCATION=${BIGQUERY_GCS_OUTPUT_LOCATION}

Đặt định dạng của tệp đầu vào.

GCS_TO_GCS_INPUT_FORMAT=csv

Đặt định dạng đầu ra mong muốn. Bạn có thể chọn parquet, json, avro hoặc csv.

GCS_TO_GCS_OUTPUT_FORMAT=parquet

Đặt chế độ đầu ra thành overwrite. Bạn có thể chọn overwrite, append, ignore hoặc errorifexists.

GCS_TO_GCS_OUTPUT_MODE=overwrite

Đặt vị trí đầu ra.

GCS_TO_GCS_OUTPUT_LOCATION=gs://${BUCKET}/GCStoGCS

Chạy mẫu

Chạy mẫu GCStoGCS.

./bin/start.sh \
-- --template=GCSTOGCS \
        --gcs.to.gcs.input.location=${GCS_TO_GCS_INPUT_LOCATION} \
        --gcs.to.gcs.input.format=${GCS_TO_GCS_INPUT_FORMAT} \
        --gcs.to.gcs.output.format=${GCS_TO_GCS_OUTPUT_FORMAT} \
        --gcs.to.gcs.output.mode=${GCS_TO_GCS_OUTPUT_MODE} \
        --gcs.to.gcs.output.location=${GCS_TO_GCS_OUTPUT_LOCATION}

Đầu ra sẽ khá ồn ào nhưng sau khoảng một phút, bạn sẽ thấy một thông báo thành công như bên dưới.

Batch [c198787ba8e94abc87e2a0778c05ec8a] submitted.
...
Batch [c198787ba8e94abc87e2a0778c05ec8a] finished.

Bạn có thể xác minh rằng các tệp được tạo bằng cách chạy lệnh sau.

gsutil ls ${GCS_TO_GCS_OUTPUT_LOCATION}

Với mẫu này, bạn cũng có thể cung cấp các truy vấn SparkSQL bằng cách truyền gcs.to.gcs.temp.view.name và gcs.to.gcs.sql.query đến mẫu, cho phép chạy truy vấn SparkSQL trên dữ liệu trước khi ghi vào GCS.

7. Dọn dẹp tài nguyên

Để tránh phát sinh các khoản phí không cần thiết cho tài khoản GCP của bạn sau khi hoàn tất lớp học lập trình này, hãy làm như sau:

1. Xoá vùng lưu trữ Cloud Storage cho môi trường mà bạn đã tạo.

gsutil rm -r gs://${BUCKET}

2. Xoá cụm Dataproc dùng cho máy chủ nhật ký liên tục.

gcloud dataproc clusters delete ${PHS_CLUSTER_NAME} \
  --region=${REGION}

3. Xoá các công việc Dataproc không dùng máy chủ. Chuyển đến Batches Console, nhấp vào ô bên cạnh mỗi công việc mà bạn muốn xoá, rồi nhấp vào XOÁ.

Nếu chỉ tạo một dự án cho lớp học lập trình này, bạn cũng có thể xoá dự án đó (không bắt buộc):

1. Trong Bảng điều khiển của GCP, hãy chuyển đến trang Dự án.
2. Trong danh sách dự án, hãy chọn dự án bạn muốn xoá rồi nhấp vào Xoá.
3. Trong hộp này, hãy nhập mã dự án rồi nhấp vào Tắt để xoá dự án.

Thận trọng: Thao tác xoá dự án sẽ gây ra những tác động sau:

• Mọi thứ trong dự án đều bị xoá. Nếu đã dùng một dự án hiện có cho hướng dẫn này, thì khi xoá dự án đó, bạn cũng sẽ xoá mọi công việc khác mà bạn đã thực hiện trong dự án.
• Mã dự án tuỳ chỉnh sẽ bị mất. Khi tạo dự án này, có thể bạn đã tạo một mã dự án tuỳ chỉnh mà bạn muốn sử dụng trong tương lai. Để giữ lại những URL sử dụng mã dự án, chẳng hạn như URL appspot.com, hãy xoá các tài nguyên đã chọn trong dự án thay vì xoá toàn bộ dự án.

8. Bước tiếp theo

Các tài nguyên sau đây cung cấp thêm những cách mà bạn có thể tận dụng Serverless Spark:

• Tìm hiểu cách điều phối quy trình công việc Dataproc không máy chủ bằng Cloud Composer.
• Tìm hiểu cách tích hợp Dataproc Serverless với các quy trình Kubeflow.