Membuat pipeline Dataflow menggunakan Python

Membuat pipeline Dataflow menggunakan Python

Dokumen ini menunjukkan cara menggunakan Apache Beam SDK untuk Python guna membuat program yang menentukan pipeline. Kemudian, Anda menjalankan pipeline menggunakan runner lokal langsung atau runner berbasis cloud seperti Dataflow. Untuk pengantar tentang pipeline WordCount, lihat video Cara menggunakan WordCount di Apache Beam.

---

Untuk mengikuti panduan langkah demi langkah untuk tugas ini langsung di Google Cloud konsol, klik Pandu saya:

Pandu saya

---

Sebelum memulai

Login ke akun Google Cloud Anda. Jika Anda baru menggunakan Google Cloud, buat akun untuk mengevaluasi performa produk kami dalam skenario dunia nyata. Pelanggan baru juga mendapatkan kredit gratis senilai $300 untuk menjalankan, menguji, dan men-deploy workload.
Instal Google Cloud CLI.
Jika Anda menggunakan penyedia identitas (IdP) eksternal, Anda harus login ke gcloud CLI dengan identitas gabungan Anda terlebih dahulu.
Untuk melakukan inisialisasi gcloud CLI, jalankan perintah berikut:
gcloud init
Buat atau pilih Google Cloud project.
Peran yang diperlukan untuk memilih atau membuat project
• Pilih project: Memilih project tidak memerlukan peran IAM tertentu—Anda dapat memilih project mana pun yang telah diberi peran.
• Membuat project: Untuk membuat project, Anda memerlukan peran Pembuat Project (roles/resourcemanager.projectCreator), yang berisi izin resourcemanager.projects.create. Pelajari cara memberikan peran.
Catatan: Jika tidak berencana untuk menyimpan resource yang Anda buat dalam prosedur ini, buatlah project, bukan memilih project yang ada. Setelah menyelesaikan langkah ini, Anda dapat menghapus project, yang menghapus semua resource yang terkait dengan project ini.
• Buat Google Cloud project:

  gcloud projects create PROJECT_ID

  Ganti PROJECT_ID dengan nama untuk Google Cloud project yang Anda buat.

• Pilih project Google Cloud yang Anda buat:

  gcloud config set project PROJECT_ID

  Ganti PROJECT_ID dengan nama project Google Cloud Anda.
Verifikasi bahwa penagihan diaktifkan untuk project Google Cloud Anda.
Aktifkan Dataflow, Compute Engine, Cloud Logging, Cloud Storage, Google Cloud Storage JSON, BigQuery, Cloud Pub/Sub, Cloud Datastore, dan Cloud Resource Manager API:
Peran yang diperlukan untuk mengaktifkan API
Untuk mengaktifkan API, Anda memerlukan peran IAM Service Usage Admin (roles/serviceusage.serviceUsageAdmin), yang berisi izin serviceusage.services.enable. Pelajari cara memberikan peran.
gcloud services enable dataflow compute_component logging storage_component storage_api bigquery pubsub datastore.googleapis.com cloudresourcemanager.googleapis.com
Buat kredensial autentikasi lokal untuk akun pengguna Anda:
gcloud auth application-default login
Jika error autentikasi ditampilkan, dan Anda menggunakan penyedia identitas (IdP) eksternal, konfirmasi bahwa Anda telah login ke gcloud CLI dengan identitas gabungan Anda.
Memberikan peran ke akun pengguna Anda. Jalankan perintah berikut satu kali untuk setiap peran IAM berikut: roles/iam.supportUser, roles/datastream.admin, roles/monitoring.metricsScopesViewer, roles/cloudaicompanion.settingsAdmin
gcloud projects add-iam-policy-binding PROJECT_ID --member="user:USER_IDENTIFIER" --role=ROLE
Ganti kode berikut:
• PROJECT_ID: Project ID Anda.
• USER_IDENTIFIER: ID untuk akun pengguna Anda. Misalnya, [e-mail].
• ROLE: Peran IAM yang Anda berikan ke akun pengguna Anda.

Instal Google Cloud CLI.

Jika Anda menggunakan penyedia identitas (IdP) eksternal, Anda harus login ke gcloud CLI dengan identitas gabungan Anda terlebih dahulu.

Untuk melakukan inisialisasi gcloud CLI, jalankan perintah berikut:

gcloud init

Buat atau pilih Google Cloud project.

Peran yang diperlukan untuk memilih atau membuat project

• Pilih project: Memilih project tidak memerlukan peran IAM tertentu—Anda dapat memilih project mana pun yang telah diberi peran.
• Membuat project: Untuk membuat project, Anda memerlukan peran Pembuat Project (roles/resourcemanager.projectCreator), yang berisi izin resourcemanager.projects.create. Pelajari cara memberikan peran.

Catatan: Jika tidak berencana untuk menyimpan resource yang Anda buat dalam prosedur ini, buatlah project, bukan memilih project yang ada. Setelah menyelesaikan langkah ini, Anda dapat menghapus project, yang menghapus semua resource yang terkait dengan project ini.

• Buat Google Cloud project:

  gcloud projects create PROJECT_ID

  Ganti PROJECT_ID dengan nama untuk Google Cloud project yang Anda buat.

• Pilih project Google Cloud yang Anda buat:

  gcloud config set project PROJECT_ID

  Ganti PROJECT_ID dengan nama project Google Cloud Anda.

Verifikasi bahwa penagihan diaktifkan untuk project Google Cloud Anda.

Aktifkan Dataflow, Compute Engine, Cloud Logging, Cloud Storage, Google Cloud Storage JSON, BigQuery, Cloud Pub/Sub, Cloud Datastore, dan Cloud Resource Manager API:

Peran yang diperlukan untuk mengaktifkan API

Untuk mengaktifkan API, Anda memerlukan peran IAM Service Usage Admin (roles/serviceusage.serviceUsageAdmin), yang berisi izin serviceusage.services.enable. Pelajari cara memberikan peran.

gcloud services enable dataflow compute_component logging storage_component storage_api bigquery pubsub datastore.googleapis.com cloudresourcemanager.googleapis.com

Buat kredensial autentikasi lokal untuk akun pengguna Anda:

gcloud auth application-default login

Jika error autentikasi ditampilkan, dan Anda menggunakan penyedia identitas (IdP) eksternal, konfirmasi bahwa Anda telah login ke gcloud CLI dengan identitas gabungan Anda.

Memberikan peran ke akun pengguna Anda. Jalankan perintah berikut satu kali untuk setiap peran IAM berikut: roles/iam.supportUser, roles/datastream.admin, roles/monitoring.metricsScopesViewer, roles/cloudaicompanion.settingsAdmin

gcloud projects add-iam-policy-binding PROJECT_ID --member="user:USER_IDENTIFIER" --role=ROLE

Ganti kode berikut:

• PROJECT_ID: Project ID Anda.
• USER_IDENTIFIER: ID untuk akun pengguna Anda. Misalnya, [e-mail].
• ROLE: Peran IAM yang Anda berikan ke akun pengguna Anda.

1. Berikan peran ke akun layanan default Compute Engine Anda. Jalankan perintah berikut satu kali untuk setiap peran IAM berikut:

   • roles/dataflow.admin
   • roles/dataflow.worker
   • roles/storage.objectAdmin

   gcloud projects add-iam-policy-binding PROJECT_ID --member="serviceAccount:[e-mail]" --role=SERVICE_ACCOUNT_ROLE

   • Ganti PROJECT_ID dengan project ID Anda.
   • Ganti PROJECT_NUMBER dengan nomor project Anda. Untuk menemukan nomor project Anda, lihat Mengidentifikasi project atau gunakan perintah gcloud projects describe.
   • Ganti SERVICE_ACCOUNT_ROLE dengan setiap peran individual.
2. Buat bucket Cloud Storage dan konfigurasikan sebagai berikut:
   • Tetapkan kelas penyimpanan ke S (Standard).
   • Tetapkan lokasi penyimpanan sebagai berikut: US (Amerika Serikat).
   • Ganti BUCKET_NAME dengan nama bucket yang unik. Jangan sertakan informasi sensitif pada nama bucket karena namespace bucket bersifat global dan dapat dilihat publik.

   gcloud storage buckets create gs://BUCKET_NAME --default-storage-class STANDARD --location US

3. Salin Google Cloud project ID dan nama bucket Cloud Storage. Anda memerlukan nilai ini nanti dalam dokumen ini.

Menyiapkan lingkungan Anda

Dataflow tidak lagi mendukung pipeline yang menggunakan Python 2. Untuk mengetahui informasi selengkapnya, lihat halaman Dukungan Python 2 di Google Cloud. Untuk mengetahui detail tentang versi yang didukung, lihat Dukungan runtime Apache Beam.

Di bagian ini, gunakan command prompt untuk menyiapkan lingkungan virtual Python yang terisolasi untuk menjalankan project pipeline Anda dengan menggunakan venv. Proses ini memungkinkan Anda mengisolasi dependensi satu project dari dependensi project lainnya.

Jika Anda belum memiliki command prompt yang tersedia, Anda dapat menggunakan Cloud Shell. Cloud Shell sudah menginstal pengelola paket untuk Python 3, sehingga Anda dapat langsung membuat lingkungan virtual.

Untuk menginstal Python, lalu membuat lingkungan virtual, ikuti langkah-langkah berikut:

1. Pastikan Anda telah menjalankan Python 3 dan pip di sistem Anda:

   python --version
   python -m pip --version

2. Jika diperlukan, instal Python 3, lalu siapkan lingkungan virtual Python: ikuti petunjuk yang diberikan di bagian Menginstal Python dan Menyiapkan venv di halaman Menyiapkan lingkungan pengembangan Python.

Setelah menyelesaikan panduan memulai cepat, Anda dapat menonaktifkan lingkungan virtual dengan menjalankan deactivate.

Mendapatkan Apache Beam SDK

Apache Beam SDK adalah model pemrograman open source untuk pipeline data. Anda menentukan pipeline dengan program Apache Beam, lalu memilih runner, seperti Dataflow, untuk menjalankan pipeline Anda.

Untuk mendownload dan menginstal Apache Beam SDK, ikuti langkah-langkah berikut:

1. Pastikan Anda berada di lingkungan virtual Python yang Anda buat di bagian sebelumnya. Pastikan perintah dimulai dengan <env_name>, dengan env_name adalah nama lingkungan virtual.
2. Instal versi terbaru Apache Beam SDK untuk Python:

pip install apache-beam[gcp]

Menjalankan pipeline secara lokal

Untuk melihat cara menjalankan pipeline secara lokal, gunakan modul Python siap pakai untuk contoh wordcount yang disertakan dengan paket apache_beam.

Contoh pipeline wordcount melakukan hal berikut:

1. Menggunakan file teks sebagai input.

   File teks ini berada di bucket Cloud Storage dengan nama resource gs://dataflow-samples/shakespeare/kinglear.txt.

2. Mengurai setiap baris menjadi kata-kata.
3. Melakukan penghitungan frekuensi pada kata yang telah di-tokenisasi.

Untuk menyiapkan pipeline wordcount secara lokal, ikuti langkah-langkah berikut:

1. Dari terminal lokal, jalankan contoh wordcount:

   python -m apache_beam.examples.wordcount \
     --output outputs

2. Lihat output pipeline:

   more outputs*

3. Untuk keluar, tekan q.

Dengan menjalankan pipeline secara lokal, Anda dapat menguji dan men-debug program Apache Beam. Anda dapat melihat kode sumber wordcount.py di GitHub Apache Beam.

Menjalankan pipeline di layanan Dataflow

Di bagian ini, jalankan contoh pipeline wordcount dari paket apache_beam di layanan Dataflow. Contoh ini menentukan DataflowRunner sebagai parameter untuk --runner.

• Menjalankan pipeline:

  python -m apache_beam.examples.wordcount \
      --region DATAFLOW_REGION \
      --input gs://dataflow-samples/shakespeare/kinglear.txt \
      --output gs://BUCKET_NAME/results/outputs \
      --runner DataflowRunner \
      --project PROJECT_ID \
      --temp_location gs://BUCKET_NAME/tmp/

  Ganti kode berikut:

  • DATAFLOW_REGION: region tempat Anda ingin men-deploy tugas Dataflow—misalnya, europe-west1

    Flag --region menggantikan region default yang ditetapkan di server metadata, klien lokal, atau variabel lingkungan.

  • BUCKET_NAME: nama bucket Cloud Storage yang Anda salin sebelumnya
  • PROJECT_ID: Google Cloud project ID yang Anda salin sebelumnya

Catatan: Untuk menentukan akun layanan worker yang dikelola pengguna, sertakan opsi pipeline --service_account_email. Akun layanan worker yang dikelola pengguna direkomendasikan untuk workload produksi. Jika Anda tidak menentukan akun layanan worker saat membuat tugas, Dataflow akan menggunakan akun layanan default Compute Engine.

Melihat hasil penelusuran Anda

Saat Anda menjalankan pipeline menggunakan Dataflow, hasilnya akan disimpan di bucket Cloud Storage. Di bagian ini, verifikasi bahwa pipeline sedang berjalan menggunakan konsol Google Cloud atau terminal lokal.

KonsolGoogle Cloud

Untuk melihat hasil di konsol Google Cloud , ikuti langkah-langkah berikut:

1. Di konsol Google Cloud , buka halaman Tugas Dataflow.

   Buka Tugas

   Halaman Tugas menampilkan detail tugas wordcount Anda, termasuk status Berjalan pada awalnya, lalu Berhasil.

2. Buka halaman Bucket Cloud Storage.

   Buka Bucket

3. Dari daftar bucket di project Anda, klik bucket penyimpanan yang Anda buat sebelumnya.

   Di direktori wordcount, file output yang dibuat oleh tugas Anda akan ditampilkan.

Terminal lokal

Lihat hasil dari terminal Anda atau menggunakan Cloud Shell.

1. Untuk mencantumkan file output, gunakan perintah gcloud storage ls:

   gcloud storage ls gs://BUCKET_NAME/results/outputs* --long

Ganti BUCKET_NAME dengan nama bucket Cloud Storage yang digunakan dalam program pipeline.

2. Untuk melihat hasil dalam file output, gunakan perintah gcloud storage cat:

   gcloud storage cat gs://BUCKET_NAME/results/outputs*

Mengubah kode pipeline

Pipeline wordcount dalam contoh sebelumnya membedakan antara kata-kata berhuruf besar dan kecil. Langkah-langkah berikut menunjukkan cara mengubah pipeline sehingga pipeline wordcount tidak peka huruf besar/kecil.

1. Di komputer lokal Anda, download salinan terbaru kode wordcount dari repositori GitHub Apache Beam.
2. Dari terminal lokal, jalankan pipeline:

   python wordcount.py --output outputs

3. Lihat hasilnya:

   more outputs*

4. Untuk keluar, tekan q.
5. Di editor pilihan Anda, buka file wordcount.py.
6. Di dalam fungsi run, periksa langkah-langkah pipeline:

   counts = (
           lines
           | 'Split' >> (beam.ParDo(WordExtractingDoFn()).with_output_types(str))
           | 'PairWithOne' >> beam.Map(lambda x: (x, 1))
           | 'GroupAndSum' >> beam.CombinePerKey(sum))

   Setelah split, baris dibagi menjadi kata-kata sebagai string.

7. Untuk mengubah string menjadi huruf kecil, ubah baris setelah split:

   counts = (
           lines
           | 'Split' >> (beam.ParDo(WordExtractingDoFn()).with_output_types(str))
           | 'lowercase' >> beam.Map(str.lower)
           | 'PairWithOne' >> beam.Map(lambda x: (x, 1))
           | 'GroupAndSum' >> beam.CombinePerKey(sum)) 

   Modifikasi ini memetakan fungsi str.lower ke setiap kata. Baris ini setara dengan beam.Map(lambda word: str.lower(word)).
8. Simpan file dan jalankan tugas wordcount yang diubah:

   python wordcount.py --output outputs

9. Lihat hasil pipeline yang diubah:

   more outputs*

10. Untuk keluar, tekan q.
11. Jalankan pipeline yang telah diubah di layanan Dataflow:

    python wordcount.py \
        --region DATAFLOW_REGION \
        --input gs://dataflow-samples/shakespeare/kinglear.txt \
        --output gs://BUCKET_NAME/results/outputs \
        --runner DataflowRunner \
        --project PROJECT_ID \
        --temp_location gs://BUCKET_NAME/tmp/

    Ganti kode berikut:

    • DATAFLOW_REGION: region tempat Anda ingin men-deploy tugas Dataflow
    • BUCKET_NAME: nama bucket Cloud Storage Anda
    • PROJECT_ID: Project ID Google Cloud Anda

Pembersihan

Agar akun Google Cloud Anda tidak dikenai biaya untuk resource yang digunakan pada halaman ini, hapus project Google Cloud yang berisi resource tersebut.

Catatan: Jika Anda mengikuti panduan memulai ini di project baru, Anda dapat menghapus project.

1. Di konsol Google Cloud , buka halaman Buckets Cloud Storage.

   Buka Buckets

2. Klik kotak centang untuk bucket yang ingin Anda dihapus.
3. Untuk menghapus bucket, klik Hapus, lalu ikuti petunjuk.

4. Jika Anda mempertahankan project, cabut peran yang Anda berikan ke akun layanan default Compute Engine. Jalankan perintah berikut satu kali untuk setiap peran IAM berikut:

   • roles/dataflow.admin
   • roles/dataflow.worker
   • roles/storage.objectAdmin

   gcloud projects remove-iam-policy-binding PROJECT_ID \
       --member=serviceAccount:[e-mail] \
       --role=SERVICE_ACCOUNT_ROLE

5. Opsional: Cabut kredensial autentikasi yang Anda buat, dan hapus file kredensial lokal.

   gcloud auth application-default revoke

6. Opsional: Cabut kredensial dari gcloud CLI.

   gcloud auth revoke

Langkah berikutnya

• Baca tentang model pemrograman Apache Beam.
• Mengembangkan pipeline secara interaktif menggunakan notebook Apache Beam.
• Pelajari cara mendesain dan membuat pipeline Anda sendiri.
• Pelajari contoh WordCount dan Game Seluler.