Menggunakan konektor Cloud Storage dengan Apache Spark

Tutorial ini menunjukkan cara menjalankan kode contoh yang menggunakan konektor Cloud Storage dengan Apache Spark.

Tujuan

Tulis tugas Spark jumlah kata sederhana di Java, Scala, atau Python, lalu jalankan tugas tersebut di cluster Dataproc.

Biaya

Dalam dokumen ini, Anda menggunakan komponen Google Cloud yang dapat ditagih berikut:

Compute Engine
Dataproc
Cloud Storage

Untuk membuat perkiraan biaya berdasarkan proyeksi penggunaan Anda, gunakan kalkulator harga. Pengguna baru Google Cloud mungkin memenuhi syarat untuk mendapatkan uji coba gratis.

Sebelum memulai

Jalankan langkah-langkah di bawah ini sebagai persiapan untuk menjalankan kode dalam tutorial ini.

Menyiapkan project. Jika perlu, siapkan project dengan Dataproc, Compute Engine, dan Cloud Storage API yang diaktifkan, serta Google Cloud CLI yang diinstal di mesin lokal Anda.
Membuat bucket Cloud Storage. Anda memerlukan Cloud Storage untuk menyimpan data tutorial. Jika belum memiliki bucket yang siap digunakan, buat bucket baru dalam project Anda.
1. Di Konsol Google Cloud, buka halaman Bucket Cloud Storage.
  Buka halaman Bucket
2. Klik Buat bucket.
3. Di halaman Buat bucket, masukkan informasi bucket Anda. Untuk melanjutkan ke langkah berikutnya, klik Lanjutkan.
  - Untuk Beri nama bucket, masukkan nama yang memenuhi persyaratan penamaan bucket.
  - Untuk Pilih tempat untuk menyimpan data, lakukan tindakan berikut:
    - Pilih opsi Jenis lokasi.
    - Pilih opsi Lokasi.
  - Untuk Memilih kelas penyimpanan default untuk data Anda, pilih kelas penyimpanan.
  - Untuk Memilih cara mengontrol akses ke objek, pilih opsi Kontrol akses.
  - Untuk Setelan lanjutan (opsional), tentukan metode enkripsi, kebijakan retensi, atau label bucket.
4. Klik Buat.
Tetapkan variabel lingkungan lokal. Tetapkan variabel lingkungan di komputer lokal Anda. Tetapkan project-id Google Cloud Anda dan nama bucket Cloud Storage yang akan Anda gunakan untuk tutorial ini. Selain itu, berikan nama dan region dari cluster Dataproc yang sudah ada atau baru. Anda dapat membuat cluster untuk digunakan dalam tutorial ini di langkah berikutnya.
```
PROJECT=project-id
```
```
BUCKET_NAME=bucket-name
```
```
CLUSTER=cluster-name
```
```
REGION=cluster-region Example: "us-central1"
```
Buat cluster Dataproc. Jalankan perintah di bawah ini untuk membuat cluster Dataproc node tunggal di zona Compute Engine yang ditentukan.
```
gcloud dataproc clusters create ${CLUSTER} \
    --project=${PROJECT} \
    --region=${REGION} \
    --single-node
```
Perintah di atas menginstal versi cluster image default. Anda dapat menggunakan flag --image-version untuk memilih versi image bagi cluster Anda. Setiap versi image menginstal komponen library Spark dan Scala versi tertentu. Jika menyiapkan tugas jumlah kata Spark di Java atau Scala, Anda akan mereferensikan versi Spark dan Scala yang terinstal di cluster saat menyiapkan paket tugas.
Menyalin data publik ke bucket Cloud Storage. Salin cuplikan teks Shakespeare data publik ke folder input di bucket Cloud Storage Anda:
```
gsutil cp gs://pub/shakespeare/rose.txt \
    gs://${BUCKET_NAME}/input/rose.txt
```
Siapkan lingkungan pengembangan Java (Apache Maven), Scala (SBT), atau Python.
Gunakan Cloud Shell. Cloud Shell mencakup alat yang digunakan dalam tutorial ini, termasuk Apache Maven, Python, serta alat command line Google Cloud CLI gcloud dan gsutil.

Menyiapkan tugas jumlah kata Spark

Pilih tab di bawah ini untuk mengikuti langkah-langkah penyiapan paket tugas atau file yang akan dikirimkan ke cluster Anda. Anda dapat mempersiapkan salah satu jenis pekerjaan berikut;

Tugas Spark di Java menggunakan Apache Maven untuk mem-build paket JAR
Tugas Spark di Scala menggunakan SBT untuk mem-build paket JAR
Tugas Spark di Python (PySpark)

Java

Salin file pom.xml ke komputer lokal Anda. File pom.xml berikut menentukan dependensi library Scala dan Spark, yang diberi cakupan provided untuk menunjukkan bahwa cluster Dataproc akan menyediakan library ini pada saat runtime. File pom.xml tidak menentukan dependensi Cloud Storage karena konektor menerapkan antarmuka HDFS standar. Saat tugas Spark mengakses file cluster Cloud Storage (file dengan URI yang dimulai dengan gs://), sistem akan otomatis menggunakan konektor Cloud Storage untuk mengakses file di Cloud Storage

Periksa versi image cluster Anda. Ganti placeholder version dalam file untuk menampilkan versi library Spark dan Scala yang digunakan oleh versi image cluster Anda. Perhatikan bahwa nomor artefak spark-core_ adalah nomor versi major.minor Scala.

<?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">
  <modelVersion>4.0.0</modelVersion>

  <groupId>dataproc.codelab</groupId>
  <artifactId>word-count</artifactId>
  <version>1.0</version>

  <properties>
    <maven.compiler.source>1.8</maven.compiler.source>
    <maven.compiler.target>1.8</maven.compiler.target>
  </properties>

  <dependencies>
    <dependency>
      <groupId>org.scala-lang</groupId>
      <artifactId>scala-library</artifactId>
      <version>Scala version, for example, 2.11.8</version>
      <scope>provided</scope>
    </dependency>
    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-core_Scala major.minor.version, for example, 2.11</artifactId>
      <version>Spark version, for example, 2.3.1</version>
      <scope>provided</scope>
    </dependency>
  </dependencies>
</project>

Salin kode WordCount.java yang tercantum di bawah ini ke komputer lokal Anda.

Buat kumpulan direktori dengan jalur src/main/java/dataproc/codelab:
```
mkdir -p src/main/java/dataproc/codelab
```
Salin WordCount.java ke komputer lokal Anda ke src/main/java/dataproc/codelab:
```
cp WordCount.java src/main/java/dataproc/codelab
```

WordCount.java adalah tugas Spark sederhana di Java yang membaca file teks dari Cloud Storage, melakukan jumlah kata, lalu menulis hasil file teks ke Cloud Storage.

package dataproc.codelab;

import java.util.Arrays;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

public class WordCount {
  public static void main(String[] args) {
    if (args.length != 2) {
      throw new IllegalArgumentException("Exactly 2 arguments are required: <inputUri> <outputUri>");
    }
    String inputPath = args[0];
    String outputPath = args[1];
    JavaSparkContext sparkContext = new JavaSparkContext(new SparkConf().setAppName("Word Count"));
    JavaRDD<String> lines = sparkContext.textFile(inputPath);
    JavaRDD<String> words = lines.flatMap(
        (String line) -> Arrays.asList(line.split(" ")).iterator()
    );
    JavaPairRDD<String, Integer> wordCounts = words.mapToPair(
        (String word) -> new Tuple2<>(word, 1)
    ).reduceByKey(
        (Integer count1, Integer count2) -> count1 + count2
    );
    wordCounts.saveAsTextFile(outputPath);
  }
}

Buat paket.
```
mvn clean package
```
Jika build berhasil, target/word-count-1.0.jar akan dibuat.

Dorong paket ke Cloud Storage.

gsutil cp target/word-count-1.0.jar \
    gs://${BUCKET_NAME}/java/word-count-1.0.jar

Scala

Salin file build.sbt ke komputer lokal Anda. File build.sbt berikut menentukan dependensi library Scala dan Spark, yang diberi cakupan provided untuk menunjukkan bahwa cluster Dataproc akan menyediakan library ini pada saat runtime. File build.sbt tidak menentukan dependensi Cloud Storage karena konektor menerapkan antarmuka HDFS standar. Saat tugas Spark mengakses file cluster Cloud Storage (file dengan URI yang dimulai dengan gs://), sistem akan otomatis menggunakan konektor Cloud Storage untuk mengakses file di Cloud Storage
Periksa versi image cluster Anda. Ganti placeholder version dalam file untuk menampilkan versi library Spark dan Scala yang digunakan oleh versi image cluster Anda.
```
scalaVersion := "Scala version, for example, 2.11.8"

name := "word-count"
organization := "dataproc.codelab"
version := "1.0"

libraryDependencies ++= Seq(
  "org.scala-lang" % "scala-library" % scalaVersion.value % "provided",
  "org.apache.spark" %% "spark-core" % "Spark version, for example, 2.3.1" % "provided"
)
```

Salin word-count.scala ke komputer lokal Anda. Ini adalah tugas Spark sederhana di Java yang membaca file teks dari Cloud Storage, melakukan jumlah kata, lalu menulis hasil file teks ke Cloud Storage.

package dataproc.codelab

import org.apache.spark.SparkContext
import org.apache.spark.SparkConf

object WordCount {
  def main(args: Array[String]) {
    if (args.length != 2) {
      throw new IllegalArgumentException(
          "Exactly 2 arguments are required: <inputPath> <outputPath>")
    }

    val inputPath = args(0)
    val outputPath = args(1)

    val sc = new SparkContext(new SparkConf().setAppName("Word Count"))
    val lines = sc.textFile(inputPath)
    val words = lines.flatMap(line => line.split(" "))
    val wordCounts = words.map(word => (word, 1)).reduceByKey(_ + _)
    wordCounts.saveAsTextFile(outputPath)
  }
}

Buat paket.
```
sbt clean package
```
Jika build berhasil, target/scala-2.11/word-count_2.11-1.0.jar akan dibuat.

Dorong paket ke Cloud Storage.

gsutil cp target/scala-2.11/word-count_2.11-1.0.jar \
    gs://${BUCKET_NAME}/scala/word-count_2.11-1.0.jar

Python

Salin word-count.py ke komputer lokal Anda. Ini adalah tugas Spark sederhana di Python menggunakan PySpark yang membaca file teks dari Cloud Storage, melakukan jumlah kata, lalu menulis hasil file teks ke Cloud Storage.

#!/usr/bin/env python

import pyspark
import sys

if len(sys.argv) != 3:
  raise Exception("Exactly 2 arguments are required: <inputUri> <outputUri>")

inputUri=sys.argv[1]
outputUri=sys.argv[2]

sc = pyspark.SparkContext()
lines = sc.textFile(sys.argv[1])
words = lines.flatMap(lambda line: line.split())
wordCounts = words.map(lambda word: (word, 1)).reduceByKey(lambda count1, count2: count1 + count2)
wordCounts.saveAsTextFile(sys.argv[2])

Mengirimkan tugas

Jalankan perintah gcloud berikut untuk mengirimkan tugas jumlah kata ke cluster Dataproc Anda.

Java

gcloud dataproc jobs submit spark \
    --cluster=${CLUSTER} \
    --class=dataproc.codelab.WordCount \
    --jars=gs://${BUCKET_NAME}/java/word-count-1.0.jar \
    --region=${REGION} \
    -- gs://${BUCKET_NAME}/input/ gs://${BUCKET_NAME}/output/

Scala

gcloud dataproc jobs submit spark \
    --cluster=${CLUSTER} \
    --class=dataproc.codelab.WordCount \
    --jars=gs://${BUCKET_NAME}/scala/word-count_2.11-1.0.jar \
    --region=${REGION} \
    -- gs://${BUCKET_NAME}/input/ gs://${BUCKET_NAME}/output/

Python

gcloud dataproc jobs submit pyspark word-count.py \
    --cluster=${CLUSTER} \
    --region=${REGION} \
    -- gs://${BUCKET_NAME}/input/ gs://${BUCKET_NAME}/output/

Melihat output

Setelah tugas selesai, jalankan perintah gsutil gcloud CLI berikut untuk melihat output jumlah kata.

gsutil cat gs://${BUCKET_NAME}/output/*

Output jumlah kata akan mirip dengan berikut ini:

(a,2)
(call,1)
(What's,1)
(sweet.,1)
(we,1)
(as,1)
(name?,1)
(any,1)
(other,1)
(rose,1)
(smell,1)
(name,1)
(would,1)
(in,1)
(which,1)
(That,1)
(By,1)

Pembersihan

Setelah menyelesaikan tutorial, Anda dapat membersihkan resource yang dibuat agar resource tersebut berhenti menggunakan kuota dan dikenai biaya. Bagian berikut menjelaskan cara menghapus atau menonaktifkan resource ini.

Menghapus project

Cara termudah untuk menghilangkan penagihan adalah dengan menghapus project yang Anda buat untuk tutorial.

Untuk menghapus project:

Perhatian: Menghapus project memiliki efek berikut:

Semua hal dalam project akan dihapus. Jika menggunakan project yang sudah ada untuk tugas dalam dokumen ini, saat Anda menghapusnya, pekerjaan lain yang telah Anda lakukan dalam project tersebut juga akan terhapus.
Project ID kustom hilang. Saat membuat project ini, Anda mungkin telah membuat project ID kustom yang ingin digunakan di masa mendatang. Untuk mempertahankan URL yang menggunakan project ID, seperti URL appspot.com, hapus resource yang dipilih di dalam project, bukan menghapus seluruh project.

Jika Anda berencana mempelajari berbagai arsitektur, tutorial, atau panduan memulai, menggunakan kembali project dapat membantu agar Anda tidak melampaui batas kuota project.

Di konsol Google Cloud, buka halaman Manage resource.
Buka Manage resource
Pada daftar project, pilih project yang ingin Anda hapus, lalu klik Delete.
Pada dialog, ketik project ID, lalu klik Shut down untuk menghapus project.

Menghapus cluster Dataproc

Daripada menghapus project, sebaiknya Anda hanya menghapus cluster dalam project.

Menghapus bucket Cloud Storage

Konsol Google Cloud

Di konsol Google Cloud, buka halaman Buckets Cloud Storage.
Buka Buckets
Klik kotak centang untuk bucket yang ingin Anda dihapus.
Untuk menghapus bucket, klik Hapus, lalu ikuti petunjuk.

Command line

gcloud storage buckets delete BUCKET_NAME

Langkah selanjutnya

Lihat Tips penyesuaian tugas Spark