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 tugas ini langsung di Google Cloud konsol, klik Pandu saya:

Pandu saya

---

Sebelum memulai

Sign in to your Google Cloud account. If you're new to Google Cloud, create an account to evaluate how our products perform in real-world scenarios. New customers also get $300 in free credits to run, test, and deploy workloads.
Install the 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
Create or select a Google Cloud project.
Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.
• Create a Google Cloud project:

  gcloud projects create PROJECT_ID

  Replace PROJECT_ID with a name for the Google Cloud project you are creating.

• Select the Google Cloud project that you created:

  gcloud config set project PROJECT_ID

  Replace PROJECT_ID with your Google Cloud project name.
Verify that billing is enabled for your Google Cloud project.
Enable the Dataflow, Compute Engine, Cloud Logging, Cloud Storage, Google Cloud Storage JSON, BigQuery, Cloud Pub/Sub, Cloud Datastore, and Cloud Resource Manager APIs:
gcloud services enable dataflow compute_component logging storage_component storage_api bigquery pubsub datastore.googleapis.com cloudresourcemanager.googleapis.com
Create local authentication credentials for your user account:
gcloud auth application-default login
If an authentication error is returned, and you are using an external identity provider (IdP), confirm that you have signed in to the gcloud CLI with your federated identity.
Grant roles to your user account. Run the following command once for each of the following IAM roles: roles/iam.serviceAccountUser
gcloud projects add-iam-policy-binding PROJECT_ID --member="user:USER_IDENTIFIER" --role=ROLE
Replace the following:
• PROJECT_ID: your project ID.
• USER_IDENTIFIER: the identifier for your user account—for example, [e-mail].
• ROLE: the IAM role that you grant to your user account.

Install the 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

Create or select a Google Cloud project.

Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

• Create a Google Cloud project:

  gcloud projects create PROJECT_ID

  Replace PROJECT_ID with a name for the Google Cloud project you are creating.

• Select the Google Cloud project that you created:

  gcloud config set project PROJECT_ID

  Replace PROJECT_ID with your Google Cloud project name.

Verify that billing is enabled for your Google Cloud project.

Enable the Dataflow, Compute Engine, Cloud Logging, Cloud Storage, Google Cloud Storage JSON, BigQuery, Cloud Pub/Sub, Cloud Datastore, and Cloud Resource Manager APIs:

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

Create local authentication credentials for your user account:

gcloud auth application-default login

If an authentication error is returned, and you are using an external identity provider (IdP), confirm that you have signed in to the gcloud CLI with your federated identity.

Grant roles to your user account. Run the following command once for each of the following IAM roles: roles/iam.serviceAccountUser

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

Replace the following:

• PROJECT_ID: your project ID.
• USER_IDENTIFIER: the identifier for your user account—for example, [e-mail].
• ROLE: the IAM role that you grant to your user account.

1. Beri 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. Create a Cloud Storage bucket and configure it as follows:
   • Set the storage class to 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

   • Salin ID project Google Cloud 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 guna 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 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 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, 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 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 Apache Beam GitHub.

   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: the 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 --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.

   Melihat hasil Anda

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

   Google Cloud console

   Untuk melihat hasil 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 pada awalnya, lalu Succeeded.

   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*

   Ubah kode pipeline

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

   1. Di komputer lokal, 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 telah diubah:

      python wordcount.py --output outputs

   9. Lihat hasil pipeline yang diubah:

      more outputs*

   10. Untuk keluar, tekan q.
   11. Jalankan pipeline yang diubah pada 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 Anda Google Cloud

   Pembersihan

   Agar tidak menimbulkan biaya pada akun Google Cloud Anda untuk resource yang digunakan di 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. Hapus bucket:

      gcloud storage buckets delete BUCKET_NAME

      Penting: Bucket Anda harus kosong sebelum Anda dapat menghapusnya.

   2. Jika Anda mempertahankan project, batalkan 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

   3. Optional: Revoke the authentication credentials that you created, and delete the local credential file.

      gcloud auth application-default revoke

   4. Optional: Revoke credentials from the 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.