Membuat pipeline Dataflow menggunakan Python

Membuat pipeline Dataflow menggunakan Python

Dalam panduan memulai ini, Anda akan mempelajari cara menggunakan Apache Beam SDK untuk Python guna membangun program yang mendefinisikan pipeline. Kemudian, Anda akan menjalankan pipeline menggunakan runner lokal langsung atau runner berbasis cloud seperti Dataflow. Untuk pengantar pipeline WordCount, lihat video Cara menggunakan WordCount di Apache Beam.

---

Jika ingin mengikuti panduan langkah demi langkah untuk tugas ini langsung di Konsol Google Cloud, klik Pandu saya:

Pandu saya

---

Sebelum memulai

1. 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.
2. Menginstal Google Cloud CLI.

3. Untuk initialize gcloud CLI, jalankan perintah berikut:

   gcloud init

4. Buat atau pilih project Google Cloud.

   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.

   • Membuat project Google Cloud:

     gcloud projects create PROJECT_ID

     Ganti PROJECT_ID dengan nama untuk project Google Cloud 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.

5. Pastikan penagihan telah diaktifkan untuk project Google Cloud Anda.

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

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

7. Buat kredensial autentikasi lokal untuk Akun Google Anda:

   gcloud auth application-default login

8. Berikan peran ke Akun Google Anda. Jalankan perintah berikut satu kali untuk setiap peran IAM berikut: roles/iam.serviceAccountUser

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

   • Ganti PROJECT_ID dengan project ID Anda.
   • Ganti EMAIL_ADDRESS dengan alamat email Anda.
   • Ganti ROLE dengan setiap peran individual.
9. Menginstal Google Cloud CLI.

10. Untuk initialize gcloud CLI, jalankan perintah berikut:

    gcloud init

11. Buat atau pilih project Google Cloud.

    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.

    • Membuat project Google Cloud:

      gcloud projects create PROJECT_ID

      Ganti PROJECT_ID dengan nama untuk project Google Cloud 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.

12. Pastikan penagihan telah diaktifkan untuk project Google Cloud Anda.

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

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

14. Buat kredensial autentikasi lokal untuk Akun Google Anda:

    gcloud auth application-default login

15. Berikan peran ke Akun Google Anda. Jalankan perintah berikut satu kali untuk setiap peran IAM berikut: roles/iam.serviceAccountUser

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

    • Ganti PROJECT_ID dengan project ID Anda.
    • Ganti EMAIL_ADDRESS dengan alamat email Anda.
    • Ganti ROLE dengan setiap peran individual.

16. Berikan peran ke akun layanan default Compute Engine Anda. Jalankan perintah berikut sekali 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 Identify projects atau gunakan perintah gcloud projects describe.
    • Ganti SERVICE_ACCOUNT_ROLE dengan setiap peran individual.
17. Buat bucket Cloud Storage dan konfigurasikan sebagai berikut:
    • Tetapkan kelas penyimpanan ke S (Standar).
    • 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

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

Menyiapkan lingkungan Anda

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

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

Jika belum memiliki command prompt yang tersedia, Anda dapat menggunakan Cloud Shell. Cloud Shell telah 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 memiliki Python 3 dan pip yang berjalan di sistem:

   python --version
   python -m pip --version
   

2. Jika perlu, instal Python 3, lalu siapkan lingkungan virtual Python: ikuti petunjuk yang diberikan di bagian Menginstal Python dan Menyiapkan venv pada halaman Menyiapkan lingkungan pengembangan Python. Jika menggunakan Python 3.10 atau yang lebih baru, Anda juga harus mengaktifkan Dataflow Runner v2. Untuk menggunakan Runner v1, gunakan Python 3.9 atau yang lebih lama.

Catatan: Cython tidak diperlukan, tetapi jika diinstal, versinya harus 0.28.1 atau yang lebih baru. Untuk memeriksa versi Cython, jalankan pip show cython.

Setelah menyelesaikan panduan memulai, 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.

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

1. Pastikan bahwa Anda berada di lingkungan virtual Python yang Anda buat di bagian sebelumnya. Pastikan prompt dimulai dengan <env_name>, dengan env_name adalah nama lingkungan virtual.
2. Instal standar pengemasan roda Python:

   pip install wheel
   

3. Instal versi terbaru Apache Beam SDK untuk Python:

pip install 'apache-beam[gcp]'

Di Microsoft Windows, gunakan perintah berikut:

pip install apache-beam[gcp]

Bergantung pada koneksi, penginstalan Anda mungkin memerlukan waktu beberapa saat.

Menjalankan pipeline secara lokal

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

Contoh pipeline wordcount melakukan hal berikut:

1. Mengambil 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.
3. Melakukan penghitungan frekuensi pada kata yang ditokenkan.

Untuk melakukan staging 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 Apache Beam GitHub.

Menjalankan pipeline di layanan Dataflow

Di bagian ini, jalankan pipeline contoh 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 Anda.

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

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

Lihat hasil Anda

Saat Anda menjalankan pipeline menggunakan Dataflow, hasilnya disimpan dalam bucket Cloud Storage. Di bagian ini, pastikan bahwa pipeline berjalan menggunakan konsol Google Cloud atau terminal lokal.

Konsol Google Cloud

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

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

   Buka Tugas

   Halaman Jobs menampilkan detail tugas wordcount Anda, termasuk status Running di awal, lalu Succeeded.

2. Buka halaman Buckets Cloud Storage.

   Buka Bucket

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

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

Terminal lokal

Lihat hasilnya dari terminal Anda atau dengan menggunakan Cloud Shell.

1. Untuk menampilkan daftar 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 hasilnya di file output, gunakan perintah gcloud storage cat:

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

Mengubah kode pipeline

Pipeline wordcount pada contoh sebelumnya membedakan antara kata-kata huruf besar dan huruf kecil. Langkah-langkah berikut menunjukkan cara memodifikasi 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 hasil:

   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 sebagai string.

7. Untuk membuat 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 telah 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: ID project Google Cloud Anda

Pembersihan

Agar tidak menimbulkan biaya pada akun Google Cloud Anda 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 Anda, cabut peran yang Anda berikan ke akun layanan default Compute Engine. Jalankan perintah berikut sekali 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 selanjutnya

• 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.