Template Pub/Sub ke Elasticsearch

Template Pub/Sub ke Elasticsearch adalah pipeline streaming yang membaca pesan dari langganan Pub/Sub, menjalankan fungsi yang ditentukan pengguna (UDF), dan menuliskannya ke Elasticsearch sebagai dokumen. Template Dataflow menggunakan fitur aliran data Elasticsearch untuk menyimpan data deret waktu di beberapa indeks sekaligus memberi Anda satu resource bernama untuk permintaan. Aliran data sangat cocok untuk log, metrik, pelacakan, dan data lain yang terus-menerus dibuat dan disimpan di Pub/Sub.

Template ini membuat aliran data bernama logs-gcp.DATASET-NAMESPACE, dengan:

  • DATASET adalah nilai parameter template dataset atau pubsub jika tidak ditentukan.
  • NAMESPACE adalah nilai parameter template namespace, atau default jika tidak ditentukan.

Persyaratan pipeline

  • Langganan Pub/Sub sumber harus ada dan pesan harus dienkode dalam format JSON yang valid.
  • Host Elasticsearch yang dapat dijangkau secara publik di Google Cloud instance atau di Elastic Cloud dengan Elasticsearch versi 7.0 atau yang lebih tinggi. Lihat Integrasi Google Cloud untuk Elastic guna mengetahui detail selengkapnya.
  • Topik Pub/Sub untuk output error.

Parameter template

Parameter yang diperlukan

  • inputSubscription: Langganan Pub/Sub untuk menggunakan input. Misalnya, projects/<PROJECT_ID>/subscriptions/<SUBSCRIPTION_NAME>.
  • errorOutputTopic: Topik output Pub/Sub untuk memublikasikan data yang gagal, dalam format projects/<PROJECT_ID>/topics/<TOPIC_NAME>.
  • connectionUrl: URL Elasticsearch dalam format https://hostname:[port]. Jika menggunakan Elastic Cloud, tentukan CloudID. Misalnya, https://elasticsearch-host:9200.
  • apiKey: Kunci API yang dienkode Base64 untuk digunakan dalam autentikasi.

Parameter opsional

  • dataset: Jenis log yang dikirim menggunakan Pub/Sub, yang memiliki dasbor siap pakai. Nilai jenis log yang diketahui adalah audit, vpcflow, dan firewall. Nilai defaultnya adalah: pubsub.
  • namespace: Pengelompokan arbitrer, seperti lingkungan (dev, prod, atau qa), tim, atau unit bisnis strategis. Nilai defaultnya adalah: default.
  • elasticsearchTemplateVersion: ID Versi Template Dataflow, biasanya ditentukan oleh Google Cloud. Nilai defaultnya adalah: 1.0.0.
  • javascriptTextTransformGcsPath: Cloud Storage URI dari file .js yang menentukan fungsi yang ditentukan pengguna (UDF) JavaScript yang akan digunakan. Contoh, gs://my-bucket/my-udfs/my_file.js.
  • javascriptTextTransformFunctionName: Nama fungsi yang ditentukan pengguna (UDF) JavaScript yang akan digunakan. Misalnya, jika kode fungsi JavaScript Anda adalah myTransform(inJson) { /*...do stuff...*/ }, maka nama fungsinya adalah myTransform. Untuk contoh UDF JavaScript, lihat Contoh UDF (https://github.com/GoogleCloudPlatform/DataflowTemplates#udf-examples).
  • javascriptTextTransformReloadIntervalMinutes: Menentukan seberapa sering UDF dimuat ulang, dalam hitungan menit. Jika nilainya lebih besar dari 0, Dataflow akan memeriksa file UDF di Cloud Storage secara berkala, dan memuat ulang UDF jika file dimodifikasi. Parameter ini memungkinkan Anda mengupdate UDF saat pipeline sedang berjalan, tanpa perlu memulai ulang tugas. Jika nilainya adalah 0, pemuatan ulang UDF akan dinonaktifkan. Nilai defaultnya adalah 0.
  • elasticsearchUsername: Nama pengguna Elasticsearch untuk autentikasi. Jika ditentukan, nilai apiKey akan diabaikan.
  • elasticsearchPassword: Sandi Elasticsearch untuk autentikasi. Jika ditentukan, nilai apiKey akan diabaikan.
  • batchSize: Ukuran tumpukan dalam jumlah dokumen. Nilai defaultnya adalah 1000.
  • batchSizeBytes: Ukuran tumpukan dalam jumlah byte. Nilai defaultnya adalah 5242880 (5 MB).
  • maxRetryAttempts: Jumlah maksimum percobaan ulang. Harus lebih besar dari nol. Nilai defaultnya adalah 5.
  • maxRetryDuration: Durasi percobaan ulang maksimum dalam milidetik. Harus lebih besar dari nol. Nilai defaultnya adalah 60000 (1 menit).
  • propertyAsIndex: Properti dalam dokumen yang diindeks yang nilainya menentukan metadata _index untuk disertakan dengan dokumen dalam permintaan massal. Diprioritaskan daripada UDF _index. Nilai defaultnya adalah none.
  • javaScriptIndexFnGcsPath: Jalur Cloud Storage ke sumber UDF JavaScript untuk fungsi yang menentukan metadata _index untuk disertakan dengan dokumen dalam permintaan massal. Nilai defaultnya adalah none.
  • javaScriptIndexFnName: Nama fungsi JavaScript UDF yang menentukan metadata _index untuk disertakan dengan dokumen dalam permintaan massal. Nilai defaultnya adalah none.
  • propertyAsId: Properti dalam dokumen yang diindeks yang nilainya menentukan metadata _id untuk disertakan dengan dokumen dalam permintaan massal. Diprioritaskan daripada UDF _id. Nilai defaultnya adalah none.
  • javaScriptIdFnGcsPath: Jalur Cloud Storage ke sumber UDF JavaScript untuk fungsi yang menentukan metadata _id untuk disertakan dengan dokumen dalam permintaan massal. Nilai defaultnya adalah none.
  • javaScriptIdFnName: Nama fungsi JavaScript UDF yang menentukan metadata _id untuk disertakan dengan dokumen dalam permintaan massal. Nilai defaultnya adalah none.
  • javaScriptTypeFnGcsPath: Jalur Cloud Storage ke sumber UDF JavaScript untuk fungsi yang menentukan metadata _type untuk disertakan dengan dokumen dalam permintaan massal. Nilai defaultnya adalah none.
  • javaScriptTypeFnName: Nama fungsi JavaScript UDF yang menentukan metadata _type untuk disertakan dengan dokumen dalam permintaan massal. Nilai defaultnya adalah none.
  • javaScriptIsDeleteFnGcsPath: Jalur Cloud Storage ke sumber UDF JavaScript untuk fungsi yang menentukan apakah akan menghapus dokumen, bukan menyisipkan atau memperbaruinya. Fungsi ini menampilkan nilai string true atau false. Nilai defaultnya adalah none.
  • javaScriptIsDeleteFnName: Nama fungsi JavaScript UDF yang menentukan apakah akan menghapus dokumen, bukan menyisipkan atau memperbaruinya. Fungsi ini menampilkan nilai string true atau false. Nilai defaultnya adalah none.
  • usePartialUpdate: Apakah akan menggunakan update parsial (memperbarui, bukan membuat atau mengindeks, yang memungkinkan dokumen parsial) dengan permintaan Elasticsearch. Nilai defaultnya adalah false.
  • bulkInsertMethod: Apakah akan menggunakan INDEX (indeks, memungkinkan upsert) atau CREATE (membuat, error pada _id duplikat) dengan permintaan massal Elasticsearch. Nilai defaultnya adalah CREATE.
  • trustSelfSignedCerts: Apakah akan mempercayai sertifikat yang ditandatangani sendiri atau tidak. Instance Elasticsearch yang diinstal mungkin memiliki sertifikat yang ditandatangani sendiri. Aktifkan parameter ini ke true untuk melewati validasi pada sertifikat SSL. (Nilai defaultnya adalah: false).
  • disableCertificateValidation: Jika true, percayai sertifikat SSL yang ditandatangani sendiri. Instance Elasticsearch mungkin memiliki sertifikat yang ditandatangani sendiri. Untuk melewati validasi sertifikat, tetapkan parameter ini ke true. Nilai defaultnya adalah false.
  • apiKeyKMSEncryptionKey: Kunci Cloud KMS untuk mendekripsi kunci API. Parameter ini diperlukan jika apiKeySource ditetapkan ke KMS. Jika parameter ini diberikan, teruskan string apiKey yang dienkripsi. Enkripsi parameter menggunakan endpoint enkripsi KMS API. Untuk kunci, gunakan format projects/<PROJECT_ID>/locations/<KEY_REGION>/keyRings/<KEY_RING>/cryptoKeys/<KMS_KEY_NAME>. Lihat: https://cloud.google.com/kms/docs/reference/rest/v1/projects.locations.keyRings.cryptoKeys/encrypt Misalnya, projects/your-project-id/locations/global/keyRings/your-keyring/cryptoKeys/your-key-name.
  • apiKeySecretId: ID secret Secret Manager untuk apiKey. Jika apiKeySource ditetapkan ke SECRET_MANAGER, berikan parameter ini. Gunakan format projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>. For example, projects/your-project-id/secrets/your-secret/versions/your-secret-version`.
  • apiKeySource: Sumber kunci API. Nilai yang diizinkan adalah PLAINTEXT, KMS, atau SECRET_MANAGER. Parameter ini diperlukan saat Anda menggunakan Secret Manager atau KMS. Jika apiKeySource ditetapkan ke KMS, apiKeyKMSEncryptionKey dan apiKey yang dienkripsi harus diberikan. Jika apiKeySource ditetapkan ke SECRET_MANAGER, apiKeySecretId harus diberikan. Jika apiKeySource ditetapkan ke PLAINTEXT, apiKey harus diberikan. Nilai defaultnya adalah: PLAINTEXT.
  • socketTimeout: Jika ditetapkan, akan menimpa waktu tunggu percobaan ulang maksimum default dan waktu tunggu soket default (30000 ms) di Elastic RestClient.

Fungsi yang ditentukan pengguna (UDF)

Template ini mendukung fungsi yang ditentukan pengguna (UDF) di beberapa titik dalam pipeline, yang dijelaskan di bawah. Untuk mengetahui informasi selengkapnya, lihat Membuat fungsi yang ditentukan pengguna (UDF) untuk template Dataflow.

Fungsi transformasi teks

Mengubah pesan Pub/Sub menjadi dokumen Elasticsearch.

Parameter template:

  • javascriptTextTransformGcsPath: Cloud Storage URI file JavaScript.
  • javascriptTextTransformFunctionName: nama fungsi JavaScript.

Spesifikasi fungsi:

  • Input: kolom data pesan Pub/Sub, diserialisasi sebagai string JSON.
  • Output: dokumen JSON yang di-string untuk disisipkan ke Elasticsearch.

Fungsi indeks

Menampilkan indeks tempat dokumen berada.

Parameter template:

  • javaScriptIndexFnGcsPath: Cloud Storage URI file JavaScript.
  • javaScriptIndexFnName: nama fungsi JavaScript.

Spesifikasi fungsi:

  • Input: dokumen Elasticsearch, diserialisasi sebagai string JSON.
  • Output: nilai kolom metadata _index dokumen.

Fungsi ID dokumen

Menampilkan ID dokumen.

Parameter template:

  • javaScriptIdFnGcsPath: Cloud Storage URI file JavaScript.
  • javaScriptIdFnName: nama fungsi JavaScript.

Spesifikasi fungsi:

  • Input: dokumen Elasticsearch, diserialisasi sebagai string JSON.
  • Output: nilai kolom metadata _id dokumen.

Fungsi penghapusan dokumen

Menentukan apakah akan menghapus dokumen. Untuk menggunakan fungsi ini, tetapkan mode penyisipan massal ke INDEX dan berikan fungsi ID dokumen.

Parameter template:

  • javaScriptIsDeleteFnGcsPath: Cloud Storage URI file JavaScript.
  • javaScriptIsDeleteFnName: nama fungsi JavaScript.

Spesifikasi fungsi:

  • Input: dokumen Elasticsearch, diserialisasi sebagai string JSON.
  • Output: menampilkan string "true" untuk menghapus dokumen, atau "false" untuk melakukan upsert dokumen.

Fungsi jenis pemetaan

Menampilkan jenis pemetaan dokumen.

Parameter template:

  • javaScriptTypeFnGcsPath: Cloud Storage URI file JavaScript.
  • javaScriptTypeFnName: nama fungsi JavaScript.

Spesifikasi fungsi:

  • Input: dokumen Elasticsearch, diserialisasi sebagai string JSON.
  • Output: nilai kolom metadata _type dokumen.

Menjalankan template

Konsol

  1. Buka halaman Dataflow Membuat tugas dari template.
    2. Buka Membuat tugas dari template
  2. Di kolom Nama tugas, masukkan nama tugas yang unik.
  3. Opsional: Untuk Endpoint regional, pilih nilai dari menu drop-down. Region default-nya adalah us-central1.

    Untuk mengetahui daftar region tempat Anda dapat menjalankan tugas Dataflow, lihat Lokasi Dataflow.

  4. Dari menu drop-down Template Dataflow, pilih template Pub/Sub ke Elasticsearch.
  5. Di kolom parameter yang disediakan, masukkan nilai parameter Anda.
  6. Klik Jalankan tugas.

gcloud

Di shell atau terminal Anda, jalankan template:

gcloud dataflow flex-template run JOB_NAME \
    --project=PROJECT_ID \
    --region=REGION_NAME \
    --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/PubSub_to_Elasticsearch_Flex \
    --parameters \
inputSubscription=SUBSCRIPTION_NAME,\
connectionUrl=CONNECTION_URL,\
dataset=DATASET,\
namespace=NAMESPACE,\
apiKey=APIKEY,\
errorOutputTopic=ERROR_OUTPUT_TOPIC

Ganti kode berikut:

  • PROJECT_ID: Google Cloud Project ID tempat Anda ingin menjalankan tugas Dataflow
  • JOB_NAME: nama tugas unik pilihan Anda
  • REGION_NAME: region tempat Anda ingin men-deploy tugas Dataflow—misalnya, us-central1
  • VERSION: versi template yang ingin Anda gunakan

    Anda dapat menggunakan nilai berikut:

  • ERROR_OUTPUT_TOPIC: topik Pub/Sub Anda untuk output error
  • SUBSCRIPTION_NAME: nama langganan Pub/Sub Anda
  • CONNECTION_URL: URL Elasticsearch Anda
  • DATASET: jenis log Anda
  • NAMESPACE: namespace Anda untuk set data
  • APIKEY: kunci API Anda yang dienkode base64 untuk autentikasi

API

Untuk menjalankan template menggunakan REST API, kirim permintaan POST HTTP. Untuk mengetahui informasi selengkapnya tentang API dan cakupan otorisasinya, lihat projects.templates.launch.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
{
   "launch_parameter": {
      "jobName": "JOB_NAME",
      "parameters": {
          "inputSubscription": "SUBSCRIPTION_NAME",
          "connectionUrl": "CONNECTION_URL",
          "dataset": "DATASET",
          "namespace": "NAMESPACE",
          "apiKey": "APIKEY",
          "errorOutputTopic": "ERROR_OUTPUT_TOPIC"
      },
      "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/PubSub_to_Elasticsearch_Flex",
   }
}

Ganti kode berikut:

  • PROJECT_ID: Google Cloud Project ID tempat Anda ingin menjalankan tugas Dataflow
  • JOB_NAME: nama tugas unik pilihan Anda
  • LOCATION: region tempat Anda ingin men-deploy tugas Dataflow—misalnya, us-central1
  • VERSION: versi template yang ingin Anda gunakan

    Anda dapat menggunakan nilai berikut:

  • ERROR_OUTPUT_TOPIC: topik Pub/Sub Anda untuk output error
  • SUBSCRIPTION_NAME: nama langganan Pub/Sub Anda
  • CONNECTION_URL: URL Elasticsearch Anda
  • DATASET: jenis log Anda
  • NAMESPACE: namespace Anda untuk set data
  • APIKEY: kunci API Anda yang dienkode base64 untuk autentikasi

Langkah berikutnya