Memecahkan masalah error Dataflow kehabisan memori

Halaman ini menjelaskan cara menemukan dan mengatasi error kehabisan memori (OOM) di Dataflow.

Menemukan error kehabisan memori

Untuk menentukan apakah pipeline Anda kehabisan memori, gunakan salah satu metode berikut.

• Di halaman Detail tugas, di panel Log , lihat tab Diagnostik. Tab ini menampilkan error terkait masalah memori dan frekuensi terjadinya error.
• Di antarmuka pemantauan Dataflow, gunakan diagram Penggunaan memori untuk memantau kapasitas dan penggunaan memori pekerja.
• Di halaman Detail tugas, di panel Log, pilih Log pekerja untuk menemukan error kehabisan memori di log pekerja.

• Error kehabisan memori mungkin juga muncul di log sistem. Untuk melihatnya, buka Logs Explorer dan gunakan kueri berikut:

  resource.type="dataflow_step"
  resource.labels.job_id="JOB_ID"
  "out of memory" OR "OutOfMemory" OR "Shutting down JVM"
  

  Ganti JOB_ID dengan ID tugas Anda.

• Untuk tugas Java, Java Memory Monitor secara berkala melaporkan metrik pengumpulan sampah. Jika fraksi waktu CPU yang digunakan untuk pengumpulan sampah melebihi nilai minimum 50% dalam jangka waktu yang lama, harness SDK akan gagal. Anda mungkin melihat error yang mirip dengan contoh berikut:

  Shutting down JVM after 8 consecutive periods of measured GC thrashing. Memory is used/total/max = ...
  

  Error ini dapat terjadi saat memori fisik masih tersedia, dan biasanya menunjukkan bahwa penggunaan memori pipeline tidak efisien. Untuk mengatasi masalah ini, optimalkan pipeline Anda.

  Java Memory Monitor dikonfigurasi oleh antarmuka MemoryMonitorOptions.

Jika tugas Anda memiliki penggunaan memori yang tinggi atau error kehabisan memori, ikuti rekomendasi di halaman ini untuk mengoptimalkan penggunaan memori atau meningkatkan jumlah memori yang tersedia.

Mengatasi error kehabisan memori

Perubahan pada pipeline Dataflow Anda dapat mengatasi error kehabisan memori atau mengurangi penggunaan memori. Perubahan yang mungkin dilakukan mencakup tindakan berikut:

• Mengoptimalkan pipeline Anda
• Mengurangi jumlah thread
• Menggunakan jenis mesin dengan lebih banyak memori per vCPU

Diagram berikut menunjukkan alur kerja pemecahan masalah Dataflow yang dijelaskan di halaman ini.

Coba mitigasi berikut:

• Jika memungkinkan, optimalkan pipeline Anda untuk mengurangi penggunaan memori.
• Jika tugas adalah tugas batch, coba langkah-langkah berikut dalam urutan yang tercantum:
  1. Gunakan jenis mesin dengan lebih banyak memori per vCPU.
  2. Kurangi jumlah thread menjadi kurang dari jumlah vCPU per pekerja.
  3. Gunakan jenis mesin kustom dengan lebih banyak memori per vCPU.
• Jika tugas adalah tugas streaming yang menggunakan Python, kurangi jumlah thread menjadi kurang dari 12.
• Jika tugas adalah tugas streaming yang menggunakan Java atau Go, coba hal berikut:
  1. Kurangi jumlah thread menjadi kurang dari 500 untuk tugas Runner v2, atau kurang dari 300 untuk tugas yang tidak menggunakan Runner v2.
  2. Gunakan jenis mesin dengan lebih banyak memori.

Mengoptimalkan pipeline Anda

Beberapa operasi pipeline dapat menyebabkan error kehabisan memori. Bagian ini memberikan opsi untuk mengurangi penggunaan memori pipeline Anda. Untuk mengidentifikasi tahap pipeline yang paling banyak menggunakan memori, gunakan Cloud Profiler untuk memantau performa pipeline.

Anda dapat menggunakan praktik terbaik berikut untuk mengoptimalkan pipeline:

• Menggunakan konektor I/O bawaan Apache Beam untuk membaca file
• Mendesain ulang operasi saat menggunakan GroupByKey PTransforms
• Mengurangi data masuk dari sumber eksternal
• Membagikan objek di seluruh thread
• Menggunakan representasi elemen yang hemat memori
• Mengurangi ukuran input tambahan
• Menggunakan DoFn yang dapat dibagi Apache Beam

Menggunakan konektor I/O bawaan Apache Beam untuk membaca file

Jangan membuka file besar di dalam DoFn. Untuk membaca file, gunakan konektor I/O bawaan Apache Beam. File yang dibuka di DoFn harus sesuai dengan memori. Karena beberapa instance DoFn berjalan secara bersamaan, file besar yang dibuka di DoFn dapat menyebabkan error kehabisan memori.

Mendesain ulang operasi saat menggunakan GroupByKey PTransforms

Saat Anda menggunakan GroupByKey PTransform di Dataflow, nilai per kunci dan per jendela yang dihasilkan akan diproses pada satu thread. Karena data ini diteruskan sebagai streaming dari layanan backend Dataflow ke pekerja, data ini tidak perlu sesuai dengan memori pekerja. Namun, jika nilai dikumpulkan dalam memori, logika pemrosesan dapat menyebabkan error kehabisan memori.

Misalnya, jika Anda memiliki kunci yang berisi data untuk jendela, dan Anda menambahkan nilai kunci ke objek dalam memori, seperti daftar, error kehabisan memori mungkin terjadi. Dalam skenario ini, pekerja mungkin tidak memiliki kapasitas memori yang cukup untuk menyimpan semua objek.

Untuk mengetahui informasi selengkapnya tentang GroupByKey PTransforms, lihat dokumentasi Apache Beam Python GroupByKey dan Java GroupByKey.

Daftar berikut berisi saran untuk mendesain pipeline guna meminimalkan penggunaan memori saat menggunakan GroupByKey PTransforms.

• Untuk mengurangi jumlah data per kunci dan per jendela, hindari kunci dengan banyak nilai, yang juga dikenal sebagai kunci aktif.
• Untuk mengurangi jumlah data yang dikumpulkan per jendela, gunakan ukuran jendela yang lebih kecil.
• Jika Anda menggunakan nilai kunci di jendela untuk menghitung angka, gunakan transformasi Combine. Jangan lakukan perhitungan dalam satu instance DoFn setelah mengumpulkan nilai.
• Filter nilai atau duplikat sebelum diproses. Untuk mengetahui informasi selengkapnya, lihat dokumentasi transformasi Python Filter dan Java Filter.

Mengurangi data masuk dari sumber eksternal

Jika Anda melakukan panggilan ke API eksternal atau database untuk pengayaan data, data yang ditampilkan harus sesuai dengan memori pekerja. Jika Anda melakukan panggilan batch, sebaiknya gunakan transformasi GroupIntoBatches. Jika Anda mengalami error kehabisan memori, kurangi ukuran batch. Untuk mengetahui informasi selengkapnya tentang pengelompokan ke dalam batch, lihat dokumentasi transformasi GroupIntoBatches Python dan Java GroupIntoBatches.

Membagikan objek di seluruh thread

Membagikan objek data dalam memori di seluruh instance DoFn dapat meningkatkan efisiensi ruang dan akses. Objek data yang dibuat dengan metode DoFn apa pun, termasuk Setup, StartBundle, Process, FinishBundle, dan Teardown, dipanggil untuk setiap DoFn. Di Dataflow, setiap pekerja mungkin memiliki beberapa instance DoFn. Untuk penggunaan memori yang lebih efisien, teruskan objek data sebagai singleton untuk membagikannya di beberapa DoFn. Untuk mengetahui informasi selengkapnya, lihat postingan blog Penggunaan ulang cache di seluruh DoFns.

Menggunakan representasi elemen yang hemat memori

Evaluasi apakah Anda dapat menggunakan representasi untuk elemen PCollection yang menggunakan lebih sedikit memori. Saat menggunakan coder di pipeline, pertimbangkan tidak hanya representasi elemen PCollection yang dienkode, tetapi juga yang didekode. Matriks jarang sering kali dapat memanfaatkan jenis pengoptimalan ini.

Mengurangi ukuran input tambahan

Jika DoFn Anda menggunakan input tambahan, kurangi ukuran input tambahan. Untuk input tambahan yang merupakan kumpulan elemen, pertimbangkan untuk menggunakan tampilan yang dapat diulang, seperti AsIterable atau AsMultimap, bukan tampilan yang mewujudkan seluruh input tambahan secara bersamaan, seperti AsList.

Mengurangi jumlah thread

Anda dapat meningkatkan memori yang tersedia per thread dengan mengurangi jumlah maksimum thread yang menjalankan instance DoFn. Perubahan ini mengurangi keparalelan, tetapi menyediakan lebih banyak memori untuk setiap DoFn.

Tabel berikut menunjukkan jumlah default thread yang dibuat Dataflow:

Jenis tugas	Python SDK	Java/Go SDK
Batch	1 thread per vCPU	1 thread per vCPU
Streaming dengan Runner v2	12 thread per vCPU	500 thread per VM pekerja
Streaming tanpa Runner v2	12 thread per vCPU	300 thread per VM pekerja

Untuk mengurangi jumlah thread Apache Beam SDK, tetapkan opsi pipeline berikut:

Java

Gunakan opsi pipeline --numberOfWorkerHarnessThreads.

Python

Gunakan opsi pipeline --number_of_worker_harness_threads.

Go

Gunakan opsi pipeline --number_of_worker_harness_threads.

Untuk tugas batch, tetapkan nilai ke angka yang kurang dari jumlah vCPU.

Untuk tugas streaming, mulai dengan mengurangi nilai menjadi setengah dari nilai default. Jika langkah ini tidak mengurangi masalah, terus kurangi nilai menjadi setengahnya, dan amati hasilnya di setiap langkah. Misalnya, saat menggunakan Python, coba nilai 6, 3, dan 1.

Menggunakan jenis mesin dengan lebih banyak memori per vCPU

Untuk memilih pekerja dengan lebih banyak memori per vCPU, gunakan salah satu metode berikut.

• Gunakan jenis mesin bermemori tinggi di kelompok mesin tujuan umum keluarga. Jenis mesin bermemori tinggi memiliki memori per vCPU yang lebih tinggi daripada jenis mesin standar. Menggunakan jenis mesin bermemori tinggi akan meningkatkan memori yang tersedia untuk setiap pekerja dan memori yang tersedia per thread, karena jumlah vCPU tetap sama. Oleh karena itu, menggunakan jenis mesin bermemori tinggi dapat menjadi cara yang hemat biaya untuk memilih pekerja dengan lebih banyak memori per vCPU.
• Untuk fleksibilitas yang lebih besar saat menentukan jumlah vCPU dan jumlah memori, Anda dapat menggunakan jenis mesin kustom. Dengan jenis mesin kustom, Anda dapat meningkatkan memori dalam penambahan 256 MB. Jenis mesin ini memiliki harga yang berbeda dengan jenis mesin standar.
• Beberapa kelompok mesin memungkinkan Anda menggunakan memori tambahan jenis mesin kustom. Memori tambahan memungkinkan rasio memori per vCPU yang lebih tinggi. Biayanya lebih tinggi. Contoh jenis mesin kustom dengan memori tambahan mencakup n2-custom-1-19456-ext dan n2-custom-8-317440-ext.

Untuk menetapkan jenis pekerja, gunakan opsi pipeline berikut. Untuk mengetahui informasi selengkapnya, lihat Menetapkan opsi pipeline dan Opsi pipeline.

Java

Gunakan opsi pipeline --workerMachineType.

Python

Gunakan opsi pipeline --machine_type.

Go

Gunakan opsi pipeline --worker_machine_type.

Hanya menggunakan satu proses Apache Beam SDK

Untuk pipeline streaming Python dan pipeline Python yang menggunakan Runner v2, Anda dapat memaksa Dataflow untuk memulai hanya satu proses Apache Beam SDK per pekerja.

Sebelum mencoba opsi ini, coba atasi masalahnya terlebih dahulu menggunakan metode lain. Untuk mengonfigurasi VM pekerja Dataflow agar hanya memulai satu proses Python dalam container, gunakan opsi pipeline berikut:

--experiments=no_use_multiple_sdk_containers

Dengan konfigurasi ini, pipeline Python membuat satu proses Apache Beam SDK per pekerja. Konfigurasi ini mencegah objek dan data bersama direplikasi beberapa kali untuk setiap proses Apache Beam SDK. Namun, konfigurasi ini membatasi penggunaan resource komputasi yang tersedia di pekerja secara efisien.

Mengurangi jumlah proses Apache Beam SDK menjadi satu tidak selalu mengurangi jumlah total thread yang dimulai di pekerja. Selain itu, memiliki semua thread pada satu proses Apache Beam SDK dapat menyebabkan pemrosesan lambat atau menyebabkan pipeline macet. Oleh karena itu, Anda mungkin juga harus mengurangi jumlah thread, seperti yang dijelaskan di bagian Mengurangi jumlah thread di halaman ini.

Anda juga dapat memaksa pekerja untuk hanya menggunakan satu proses Apache Beam SDK dengan menggunakan jenis mesin yang hanya memiliki satu vCPU.

Memahami penggunaan memori Dataflow

Untuk memecahkan masalah error kehabisan memori, sebaiknya pahami cara pipeline Dataflow menggunakan memori.

Saat Dataflow menjalankan pipeline, pemrosesan akan didistribusikan ke beberapa mesin virtual (VM) Compute Engine, yang sering disebut pekerja. Pekerja memproses item kerja dari layanan Dataflow dan mendelegasikan item kerja ke proses Apache Beam SDK. Proses Apache Beam SDK membuat instance DoFn. DoFn adalah class Apache Beam SDK yang menentukan fungsi pemrosesan terdistribusi.

Dataflow meluncurkan beberapa thread di setiap pekerja, dan memori setiap pekerja dibagikan ke semua thread. Thread adalah satu tugas yang dapat dieksekusi yang berjalan dalam proses yang lebih besar. Jumlah default thread bergantung pada beberapa faktor dan bervariasi antara tugas batch dan streaming.

Jika pipeline Anda memerlukan lebih banyak memori daripada jumlah memori default yang tersedia di pekerja, Anda mungkin akan mengalami error kehabisan memori.

Pipeline Dataflow terutama menggunakan memori pekerja dengan tiga cara:

• Memori operasional pekerja
• Memori proses SDK
• DoFn penggunaan memori

Memori operasional pekerja

Pekerja Dataflow memerlukan memori untuk sistem operasi dan proses sistemnya. Penggunaan memori pekerja biasanya tidak lebih dari 1 GB. Penggunaan biasanya kurang dari 1 GB.

• Berbagai proses di pekerja menggunakan memori untuk memastikan pipeline Anda berfungsi dengan baik. Setiap proses ini mungkin mencadangkan sejumlah kecil memori untuk operasinya.
• Saat pipeline Anda tidak menggunakan Streaming Engine, proses pekerja tambahan akan menggunakan memori.

Memori proses SDK

Proses Apache Beam SDK dapat membuat objek dan data yang dibagikan antar-thread dalam proses, yang disebut di halaman ini sebagai objek dan data bersama SDK. Penggunaan memori dari objek dan data bersama SDK ini disebut sebagai memori proses SDK. Daftar berikut mencakup contoh objek dan data bersama SDK:

• Input tambahan
• Model machine learning
• Objek singleton dalam memori
• Objek Python yang dibuat dengan modul apache_beam.utils.shared
• Data yang dimuat dari sumber eksternal, seperti Cloud Storage atau BigQuery

Tugas streaming yang tidak menggunakan Streaming Engine menyimpan input tambahan dalam memori. Untuk pipeline Java dan Go, setiap pekerja memiliki satu salinan input tambahan. Untuk pipeline Python, setiap proses Apache Beam SDK memiliki satu salinan input tambahan.

Tugas streaming yang menggunakan Streaming Engine memiliki batas ukuran input tambahan sebesar 80 MB. Input tambahan disimpan di luar memori pekerja.

Penggunaan memori dari objek dan data bersama SDK tumbuh secara linear dengan jumlah proses Apache Beam SDK. Di pipeline Java dan Go, satu proses Apache Beam SDK dimulai per pekerja. Di pipeline Python, satu proses Apache Beam SDK dimulai per vCPU. Objek dan data bersama SDK digunakan kembali di seluruh thread dalam proses Apache Beam SDK yang sama.

Penggunaan memori DoFn

DoFn adalah class Apache Beam SDK yang menentukan fungsi pemrosesan terdistribusi. Setiap pekerja dapat menjalankan instance DoFn serentak. Setiap thread menjalankan satu instance DoFn. Saat mengevaluasi total penggunaan memori, menghitung ukuran set kerja, atau jumlah memori yang diperlukan agar aplikasi dapat terus berfungsi, mungkin akan membantu. Misalnya, jika satu DoFn menggunakan memori maksimum 5 MB dan pekerja memiliki 300 thread, penggunaan memori DoFn dapat mencapai 1,5 GB, atau jumlah byte memori dikalikan dengan jumlah thread. Bergantung pada cara pekerja menggunakan memori, lonjakan penggunaan memori dapat menyebabkan pekerja kehabisan memori.

Sulit untuk memperkirakan jumlah instance DoFn yang dibuat Dataflow. Jumlahnya bergantung pada berbagai faktor, seperti SDK, jenis mesin, dan variabel lainnya. Selain itu, DoFn mungkin digunakan oleh beberapa thread secara berurutan. Layanan Dataflow tidak menjamin berapa kali DoFn dipanggil, dan tidak menjamin jumlah pasti instance DoFn yang dibuat selama pipeline. Namun, tabel berikut memberikan beberapa insight tentang tingkat keparalelan yang dapat Anda harapkan dan memperkirakan batas atas jumlah instance DoFn.

Beam Python SDK

	Batch	Streaming tanpa Streaming Engine	Streaming Engine
Keparalelan	1 proses per vCPU 1 thread per proses 1 thread per vCPU	1 proses per vCPU 12 thread per proses 12 thread per vCPU	1 proses per vCPU 12 thread per proses 12 thread per vCPU
Jumlah maksimum instance DoFn serentak (Semua angka ini dapat berubah kapan saja.)	1 DoFn per thread 1 DoFn per vCPU	1 DoFn per thread 12 DoFn per vCPU	1 DoFn per thread 12 DoFn per vCPU

Beam Java/Go SDK

	Batch	Streaming Appliance dan Streaming Engine tanpa runner v2	Streaming Engine dengan runner v2
Keparalelan	1 proses per VM pekerja 1 thread per vCPU	1 proses per VM pekerja 300 thread per proses 300 thread per VM pekerja	1 proses per VM pekerja 500 thread per proses 500 thread per VM pekerja
Jumlah maksimum instance DoFn serentak (Semua angka ini dapat berubah kapan saja.)	1 DoFn per thread 1 DoFn per vCPU	1 DoFn per thread 300 DoFn per VM pekerja	1 DoFn per thread 500 DoFn per VM pekerja

Misalnya, saat menggunakan Python SDK dengan pekerja Dataflow n1-standard-2, hal berikut berlaku:

• Tugas batch: Dataflow meluncurkan satu proses per vCPU (dua dalam kasus ini). Setiap proses menggunakan satu thread, dan setiap thread membuat satu instance DoFn.
• Tugas streaming dengan Streaming Engine: Dataflow memulai satu proses per vCPU (total dua). Namun, setiap proses dapat membuat hingga 12 thread, masing-masing dengan instance DoFn-nya sendiri.

Saat mendesain pipeline yang kompleks, penting untuk memahami DoFn siklus proses. Pastikan fungsi DoFn Anda dapat diserialisasi, dan hindari mengubah argumen elemen secara langsung di dalamnya.

Jika Anda memiliki pipeline multi-bahasa, dan lebih dari satu Apache Beam SDK berjalan di pekerja, pekerja akan menggunakan tingkat keparalelan thread per proses terendah yang memungkinkan.

Perbedaan Java, Go, dan Python

Java, Go, dan Python mengelola proses dan memori secara berbeda. Oleh karena itu, pendekatan yang harus Anda ambil saat memecahkan masalah error kehabisan memori bervariasi berdasarkan apakah pipeline Anda menggunakan Java, Go, atau Python.

Pipeline Java dan Go

Di pipeline Java dan Go:

• Setiap pekerja memulai satu proses Apache Beam SDK.
• Objek dan data bersama SDK, seperti input tambahan dan cache, dibagikan di antara semua thread di pekerja.
• Memori yang digunakan oleh objek dan data bersama SDK biasanya tidak diskalakan berdasarkan jumlah vCPU di pekerja.

Pipeline Python

Di pipeline Python:

• Setiap pekerja memulai satu proses Apache Beam SDK per vCPU.
• Objek dan data bersama SDK, seperti input tambahan dan cache, dibagikan di antara semua thread dalam setiap proses Apache Beam SDK.
• Jumlah total thread di pekerja diskalakan secara linear berdasarkan jumlah vCPU. Oleh karena itu, memori yang digunakan oleh objek dan data bersama SDK tumbuh secara linear dengan jumlah vCPU.
• Thread yang melakukan pekerjaan didistribusikan di seluruh proses. Unit kerja baru ditetapkan ke proses tanpa item kerja, atau ke proses dengan item kerja paling sedikit yang ditetapkan pada saat itu.