1. Genel bakış - Serverless for Apache Spark
Managed Service for Apache Spark; Apache Spark, Apache Flink, Presto ve diğer birçok açık kaynak aracı ve çerçevesini çalıştırmak için geliştirilmiş, tümüyle yönetilen ve yüksek oranda ölçeklenebilir bir hizmettir. Gezegen ölçeğinde veri gölü modernizasyonu, ETL / ELT ve güvenli veri bilimi için Managed Apache Spark'ı kullanın. Managed Apache Spark; BigQuery, Cloud Storage, Gemini Enterprise Agent Engine ve Knowledge Catalog dahil olmak üzere çeşitli Google Cloud hizmetleriyle de tamamen entegre edilmiştir.
Yönetilen Apache Spark iki farklı seçenek olarak sunulur:
- Yönetilen Apache Spark sunucusuz, altyapı ve otomatik ölçeklendirme yapılandırmanıza gerek kalmadan PySpark işlerini çalıştırmanıza olanak tanır. Yönetilen Apache Spark, PySpark toplu iş yüklerini ve oturumlarını / not defterlerini destekler.
- Yönetilen Apache Spark kümeleri, Flink ve Presto gibi açık kaynak araçların yanı sıra YARN tabanlı Spark iş yükleri için bir Hadoop YARN kümesini yönetmenize olanak tanır. Otomatik ölçeklendirme de dahil olmak üzere, bulut tabanlı kümelerinizi istediğiniz kadar dikey veya yatay ölçeklendirme ile özelleştirebilirsiniz.
Bu codelab'de, Dataproc Serverless'ı kullanabileceğiniz çeşitli yöntemleri öğreneceksiniz.
Apache Spark, başlangıçta Hadoop kümelerinde çalışacak şekilde oluşturulmuştu ve kaynak yöneticisi olarak YARN'ı kullanıyordu. Hadoop kümelerinin bakımı için belirli bir uzmanlık ve kümelerdeki birçok farklı ayarın düzgün şekilde yapılandırılması gerekir. Bu, Spark'ın kullanıcının ayarlamasını gerektirdiği ayrı bir düğme grubuna ek olarak sunulur. Bu durum, geliştiricilerin Spark kodu üzerinde çalışmak yerine altyapılarını yapılandırmak için daha fazla zaman harcadığı birçok senaryoya yol açar.
Dataproc Serverless, Hadoop kümelerini veya Spark'ı manuel olarak yapılandırma ihtiyacını ortadan kaldırır. Dataproc Serverless, Hadoop üzerinde çalışmaz ve otomatik ölçeklendirme dahil olmak üzere kaynak gereksinimlerini belirlemek için kendi Dinamik Kaynak Ayırma özelliğini kullanır. Spark özelliklerinin küçük bir alt kümesi Dataproc Serverless ile özelleştirilebilir ancak çoğu durumda bunları ayarlamanız gerekmez.
2. Kur
Öncelikle ortamınızı ve bu codelab'de kullanılan kaynakları yapılandıracaksınız.
Google Cloud projesi oluşturun. Mevcut bir hesabı kullanabilirsiniz.
Cloud Console araç çubuğunda Cloud Shell'i tıklayarak açın.
Cloud Shell, bu codelab'de kullanabileceğiniz, kullanıma hazır bir kabuk ortamı sağlar.
Cloud Shell, proje adınızı varsayılan olarak ayarlar. echo $GOOGLE_CLOUD_PROJECT komutunu çalıştırarak tekrar kontrol edin. Çıktıda proje kimliğinizi görmüyorsanız ayarlayın.
export GOOGLE_CLOUD_PROJECT=<your-project-id>
Kaynaklarınız için us-central1 veya europe-west2 gibi bir Compute Engine bölgesi ayarlayın.
export REGION=<your-region>
API'leri etkinleştir
Bu codelab'de aşağıdaki API'ler kullanılır:
- BigQuery
- Dataproc
Gerekli API'leri etkinleştirin. Bu işlem yaklaşık bir dakika sürer ve tamamlandığında işlemin başarılı olduğunu belirten bir mesaj görüntülenir.
gcloud services enable bigquery.googleapis.com gcloud services enable dataproc.googleapis.com
Ağ erişimini yapılandırma
Spark sürücüleri ve yürütücüleri yalnızca özel IP'lere sahip olduğundan Dataproc Serverless, Spark işlerinizi çalıştıracağınız bölgede Google Özel Erişimi'nin etkinleştirilmesini gerektirir. default alt ağında etkinleştirmek için aşağıdakileri çalıştırın.
gcloud compute networks subnets update default \
--region=${REGION} \
--enable-private-ip-google-access
Aşağıdaki komutları kullanarak Google Özel Erişim'in etkinleştirildiğini doğrulayabilirsiniz. Bu komutlar True veya False değerini verir.
gcloud compute networks subnets describe default \
--region=${REGION} \
--format="get(privateIpGoogleAccess)"
Storage paketi oluşturma
Bu codelab'de oluşturulan öğeleri depolamak için kullanılacak bir depolama paketi oluşturun.
Paketiniz için bir ad seçin. Paket adları tüm kullanıcılar arasında genel olarak benzersiz olmalıdır.
export BUCKET=<your-bucket-name>
Paketi, Spark işlerinizi çalıştırmayı planladığınız bölgede oluşturun.
gsutil mb -l ${REGION} gs://${BUCKET}
Paketinizin Cloud Storage konsolunda kullanılabildiğini görebilirsiniz. Ayrıca, bucket'ınızı görmek için gsutil ls komutunu da çalıştırabilirsiniz.
Kalıcı İş Geçmişi Sunucusu oluşturma
Spark kullanıcı arayüzü, zengin bir hata ayıklama araçları seti ve Spark işleriyle ilgili analizler sunar. Tamamlanan Dataproc Serverless işleri için Spark kullanıcı arayüzünü görüntülemek üzere kalıcı geçmiş sunucusu olarak kullanılacak tek düğümlü bir Dataproc kümesi oluşturmanız gerekir.
Kalıcı geçmiş sunucunuz için bir ad belirleyin.
PHS_CLUSTER_NAME=my-phs
Aşağıdaki komutu çalıştırın.
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
Spark kullanıcı arayüzü ve kalıcı geçmiş sunucusu, bu codelab'in ilerleyen bölümlerinde daha ayrıntılı olarak ele alınacaktır.
3. Dataproc Batches ile sunucusuz Spark işleri çalıştırma
Bu örnekte, New York City (NYC) Citi Bike Trips herkese açık veri kümesindeki bir veri grubuyla çalışacaksınız. NYC Citi Bikes, NYC'de ücretli bir bisiklet paylaşım sistemidir. Bazı basit dönüşümler gerçekleştirecek ve en popüler on Citi Bike istasyon kimliğini yazdıracaksınız. Bu örnekte, Spark ile BigQuery arasında verileri sorunsuz bir şekilde okuyup yazmak için açık kaynaklı spark-bigquery-connector da kullanılıyor.
Aşağıdaki GitHub deposunu klonlayın ve cd dosyasını citibike.py dosyasını içeren dizine kopyalayın.
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")
Varsayılan olarak Cloud Shell'de bulunan Cloud SDK'yı kullanarak işi sunucusuz Spark'a gönderin. Sunucusuz Spark işlerini göndermek için Cloud SDK'yı ve Dataproc Batches API'yi kullanan kabuğunuzda aşağıdaki komutu çalıştırın.
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}
Bu durumu daha iyi anlamak için:
gcloud dataproc batches submit, Dataproc Batches API'yi ifade eder.pyspark, PySpark işi gönderdiğinizi gösterir.--batch, işin adıdır. Sağlanmazsa rastgele oluşturulan bir UUID kullanılır.--region=${REGION}, işin işleneceği coğrafi bölgedir.--deps-bucket=${BUCKET}, yerel Python dosyanızın sunucusuz ortamda çalıştırılmadan önce yüklendiği yerdir.--jars=gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.26.0.jar, Spark çalışma zamanı ortamında spark-bigquery-connector için JAR dosyasını içerir.--history-server-cluster=projects/${GOOGLE_CLOUD_PROJECT}/regions/${REGION}/clusters/${PHS_CLUSTER}, kalıcı geçmiş sunucusunun tam nitelikli adıdır. Spark etkinlik verileri (konsol çıkışından ayrı) burada depolanır ve Spark kullanıcı arayüzünden görüntülenebilir.- Sondaki
--, bundan sonraki her şeyin program için çalışma zamanı bağımsız değişkenleri olacağını gösterir. Bu durumda, görevde belirtildiği şekilde paketinizin adını gönderiyorsunuz.
Toplu işlem gönderildiğinde aşağıdaki çıkışı görürsünüz.
Batch [citibike-job] submitted.
Birkaç dakika sonra, işin meta verileriyle birlikte aşağıdaki çıkışı görürsünüz.
+----------------+------+ |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.
Sonraki bölümde, bu işin günlüklerini nasıl bulacağınızı öğreneceksiniz.
Ek özellikler
Spark Serverless ile işlerinizi çalıştırmak için ek seçenekleriniz olur.
- İşinizin üzerinde çalışacağı bir özel Docker görüntüsü oluşturabilirsiniz. Bu, Python ve R kitaplıkları da dahil olmak üzere ek bağımlılıkları dahil etmenin harika bir yoludur.
- Hive meta verilerine erişmek için işinize bir Dataproc Metastore örneği bağlayabilirsiniz.
- Dataproc Serverless, daha fazla kontrol için küçük bir Spark özelliği grubunun yapılandırılmasını destekler.
4. Dataproc Metrikleri ve Gözlemlenebilirlik
Dataproc Batches Console, tüm Dataproc Serverless işlerinizi listeler. Konsolda her işin Toplu İşlem Kimliği, Konum, Durum, Oluşturma Zamanı, Geçen Süre ve Tür bilgilerini görürsünüz. İşinizle ilgili daha fazla bilgi görüntülemek için Toplu İş Kimliği'ni tıklayın.
Bu sayfada, işinizin zaman içinde kaç Batch Spark Executor kullandığını (ne kadar otomatik ölçeklendiğini gösterir) gösteren İzleme gibi bilgiler görürsünüz.
Ayrıntılar sekmesinde, işin gönderilmesiyle birlikte gönderilen tüm bağımsız değişkenler ve parametreler de dahil olmak üzere işle ilgili daha fazla meta veri görürsünüz.
Tüm günlükleri bu sayfadan da görüntüleyebilirsiniz. Dataproc Serverless işleri çalıştırıldığında üç farklı günlük grubu oluşturulur:
- Hizmet düzeyi
- Konsol çıktısı
- Spark etkinliklerinin günlüğe kaydedilmesi
Hizmet düzeyi: Dataproc Serverless hizmetinin oluşturduğu günlükleri içerir. Bunlar arasında, Dataproc Serverless'ın otomatik ölçeklendirme için ek CPU istemesi gibi işlemler yer alır. Bu günlükleri görüntülemek için Günlükleri görüntüle'yi tıklayın. Bu işlem, Cloud Logging'i açar.
Konsol çıkışı, Çıkış bölümünde görüntülenebilir. Bu, iş tarafından oluşturulan çıktıdır. Buna, Spark'ın bir işe başlarken yazdırdığı meta veriler veya işe dahil edilen yazdırma ifadeleri dahildir.
Spark etkinlik günlüğü, Spark kullanıcı arayüzünden erişilebilir. Spark işinize kalıcı bir geçmiş sunucusu sağladığınız için, daha önce çalıştırdığınız Spark işleriyle ilgili bilgileri içeren View Spark History Server'ı (Spark Geçmiş Sunucusunu Görüntüle) tıklayarak Spark kullanıcı arayüzüne erişebilirsiniz. Spark kullanıcı arayüzü hakkında daha fazla bilgiyi resmi Spark belgelerinden edinebilirsiniz.
5. Dataproc Şablonları: BQ -> GCS
Dataproc şablonları, buluttaki veri işleme görevlerini daha da basitleştirmeye yardımcı olan açık kaynak araçlardır. Bunlar, Dataproc Sunucusuz için sarmalayıcı görevi görür ve aşağıdakiler de dahil olmak üzere birçok veri içe ve dışa aktarma görevi için şablonlar içerir:
BigQuerytoGCSveGCStoBigQueryGCStoBigTableGCStoJDBCveJDBCtoGCSHivetoBigQueryMongotoGCSveGCStoMongo
Tam listeyi README dosyasında bulabilirsiniz.
Bu bölümde, BigQuery'den GCS'ye veri aktarmak için Dataproc şablonlarını kullanacaksınız.
Depoyu klonlama
Depoyu klonlayın ve python klasörüne geçin.
git clone https://github.com/GoogleCloudPlatform/dataproc-templates.git cd dataproc-templates/python
Ortamı yapılandırma
Şimdi ortam değişkenlerini ayarlayacaksınız. Dataproc şablonları, proje kimliğiniz için GCP_PROJECT ortam değişkenini kullanır. Bu nedenle, bu değişkeni GOOGLE_CLOUD_PROJECT. olarak ayarlayın.
export GCP_PROJECT=${GOOGLE_CLOUD_PROJECT}
Bölgeniz, önceki ortamda ayarlanmış olmalıdır. Değilse buradan ayarlayın.
export REGION=<region>
Dataproc şablonları, BigQuery işlerini işlemek için spark-bigquery-conector'ı kullanır ve URI'nin JARS ortam değişkenine dahil edilmesi gerekir. JARS değişkenini ayarlayın.
export JARS="gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.26.0.jar"
Şablon parametrelerini yapılandırma
Hizmetin kullanacağı bir hazırlama paketi adı ayarlayın.
export GCS_STAGING_LOCATION=gs://${BUCKET}
Ardından, işe özel bazı değişkenleri ayarlayacaksınız. Giriş tablosu için yine BigQuery NYC Citibike veri kümesine başvuracaksınız.
BIGQUERY_GCS_INPUT_TABLE=bigquery-public-data.new_york_citibike.citibike_trips
csv, parquet, avro veya json seçeneklerinden birini belirleyebilirsiniz. Bu codelab için CSV'yi seçin. Bir sonraki bölümde, dosya türlerini dönüştürmek için Dataproc şablonlarının nasıl kullanılacağı açıklanmaktadır.
BIGQUERY_GCS_OUTPUT_FORMAT=csv
Çıkış modunu overwrite olarak ayarlayın. overwrite, append, ignore veya errorifexists. arasından seçim yapabilirsiniz.
BIGQUERY_GCS_OUTPUT_MODE=overwrite
GCS çıkış konumunu paketinize ait bir yol olarak ayarlayın.
BIGQUERY_GCS_OUTPUT_LOCATION=gs://${BUCKET}/BQtoGCS
Şablonu çalıştırma
Aşağıda belirterek ve ayarladığınız giriş parametrelerini sağlayarak BIGQUERYTOGCS şablonunu çalıştırın.
./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}
Çıkış oldukça gürültülü olur ancak yaklaşık bir dakika sonra aşağıdakileri görürsünüz.
Batch [5766411d6c78444cb5e80f305308d8f8] submitted. ... Batch [5766411d6c78444cb5e80f305308d8f8] finished.
Aşağıdaki komutu çalıştırarak dosyaların oluşturulduğunu doğrulayabilirsiniz.
gsutil ls ${BIGQUERY_GCS_OUTPUT_LOCATION}
Spark, veri miktarına bağlı olarak varsayılan olarak birden fazla dosyaya yazar. Bu durumda yaklaşık 30 oluşturulmuş dosya görürsünüz. Spark çıkış dosyası adları, part ile biçimlendirilir. Ardından beş basamaklı bir sayı (parça numarasını gösterir) ve bir karma dizesi gelir. Büyük miktarlarda veri için Spark genellikle birkaç dosyaya yazar. Örnek dosya adı: part-00000-cbf69737-867d-41cc-8a33-6521a725f7a0-c000.csv.
6. Dataproc Şablonları: CSV'den Parquet'ye
Şimdi, GCSTOGCS kullanarak GCS'deki verileri bir dosya türünden diğerine dönüştürmek için Dataproc şablonlarını kullanacaksınız. Bu şablon SparkSQL kullanır ve ek işleme için dönüşüm sırasında işlenecek bir SparkSQL sorgusu gönderme seçeneği de sunar.
Ortam değişkenlerini onaylama
GCP_PROJECT, REGION ve GCS_STAGING_BUCKET değerlerinin önceki bölümde ayarlandığını doğrulayın.
echo ${GCP_PROJECT}
echo ${REGION}
echo ${GCS_STAGING_LOCATION}
Şablon parametrelerini ayarlama
Şimdi GCStoGCS için yapılandırma parametrelerini ayarlayacaksınız. Giriş dosyalarının konumuyla başlayın. Bu işlemin, dizindeki tüm dosyalar işleneceğinden belirli bir dosya değil, bir dizin olduğunu unutmayın. Bunu BIGQUERY_GCS_OUTPUT_LOCATION olarak ayarlayın.
GCS_TO_GCS_INPUT_LOCATION=${BIGQUERY_GCS_OUTPUT_LOCATION}
Giriş dosyasının biçimini ayarlayın.
GCS_TO_GCS_INPUT_FORMAT=csv
İstediğiniz çıkış biçimini ayarlayın. Parquet, JSON, Avro veya CSV'yi seçebilirsiniz.
GCS_TO_GCS_OUTPUT_FORMAT=parquet
Çıkış modunu overwrite olarak ayarlayın. overwrite, append, ignore veya errorifexists. arasından seçim yapabilirsiniz.
GCS_TO_GCS_OUTPUT_MODE=overwrite
Çıkış konumunu ayarlayın.
GCS_TO_GCS_OUTPUT_LOCATION=gs://${BUCKET}/GCStoGCS
Şablonu çalıştırma
GCStoGCS şablonunu çalıştırın.
./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}
Çıktı oldukça gürültülü olacaktır ancak yaklaşık bir dakika sonra aşağıdakine benzer bir başarı mesajı görmeniz gerekir.
Batch [c198787ba8e94abc87e2a0778c05ec8a] submitted. ... Batch [c198787ba8e94abc87e2a0778c05ec8a] finished.
Aşağıdaki komutu çalıştırarak dosyaların oluşturulduğunu doğrulayabilirsiniz.
gsutil ls ${GCS_TO_GCS_OUTPUT_LOCATION}
Bu şablonla, gcs.to.gcs.temp.view.name ve gcs.to.gcs.sql.query değerlerini şablona ileterek SparkSQL sorguları sağlama seçeneğiniz de vardır. Bu sayede, GCS'ye yazmadan önce veriler üzerinde bir SparkSQL sorgusu çalıştırılabilir.
7. Kaynakları temizleme
Bu codelab tamamlandıktan sonra GCP hesabınızın gereksiz yere ücretlendirilmesini önlemek için:
- Oluşturduğunuz ortam için Cloud Storage paketini silin.
gsutil rm -r gs://${BUCKET}
- Kalıcı geçmiş sunucunuz için kullanılan Dataproc kümesini silin.
gcloud dataproc clusters delete ${PHS_CLUSTER_NAME} \
--region=${REGION}
- Dataproc sunucusuz işlerini silin. Toplu İşler Konsolu'na gidin, silmek istediğiniz her işin yanındaki kutuyu işaretleyin ve SİL'i tıklayın.
Yalnızca bu codelab için bir proje oluşturduysanız isteğe bağlı olarak projeyi de silebilirsiniz:
- GCP Console'da Projeler sayfasına gidin.
- Proje listesinde, silmek istediğiniz projeyi seçin ve Sil'i tıklayın.
- Kutuda proje kimliğini yazın ve projeyi silmek için Kapat'ı tıklayın.
8. Sırada ne var?
Aşağıdaki kaynaklar, Serverless Spark'tan yararlanabileceğiniz ek yöntemler sunar:
- Cloud Composer'ı kullanarak Dataproc Serverless iş akışlarını nasıl düzenleyeceğinizi öğrenin.
- Dataproc Serverless'ı Kubeflow ardışık düzenleriyle nasıl entegre edeceğinizi öğrenin.