Template Pub/Sub ke Elasticsearch

Template Pub/Sub ke Elasticsearch adalah pipeline streaming yang membaca pesan dari langganan Pub/Sub, mengeksekusi 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, rekaman aktivitas, dan data lain yang dibuat secara berkelanjutan 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 instance Google Cloud Platform atau di Elastic Cloud dengan Elasticsearch versi 7.0 atau yang lebih tinggi. Lihat Integrasi Google Cloud untuk Elastic untuk mengetahui detail selengkapnya.
  • Topik Pub/Sub untuk output error.

Parameter template

Parameter yang diperlukan

  • inputSubscription: Langganan Pub/Sub untuk menggunakan input dari. Contoh, 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 berenkode Base64 yang akan digunakan untuk 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. Default: pubsub.
  • namespace: Pengelompokan arbitrer, seperti lingkungan (dev, prod, atau qa), tim, atau unit bisnis strategis. Default: 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 mengautentikasi. Jika ditentukan, nilai apiKey akan diabaikan.
  • elasticsearchPassword: Sandi Elasticsearch untuk melakukan autentikasi. Jika ditentukan, nilai apiKey akan diabaikan.
  • batchSize: Ukuran batch dalam jumlah dokumen. Nilai defaultnya adalah 1000.
  • batchSizeBytes: Ukuran batch dalam jumlah byte. Nilai defaultnya adalah 5242880 (5 MB).
  • maxRetryAttempts: Jumlah maksimum upaya percobaan ulang. Harus lebih besar dari nol. Nilai defaultnya adalah no retries.
  • maxRetryDuration: Durasi percobaan ulang maksimum dalam milidetik. Harus lebih besar dari nol. Nilai defaultnya adalah no retries.
  • propertyAsIndex: Properti dalam dokumen yang diindeks yang nilainya menentukan metadata _index untuk disertakan dengan dokumen dalam permintaan massal. Lebih diprioritaskan daripada UDF _index. Setelan defaultnya adalah none.
  • javaScriptIndexFnGcsPath: Jalur Cloud Storage ke sumber UDF JavaScript untuk fungsi yang menentukan metadata _index yang akan disertakan dengan dokumen dalam permintaan massal. Setelan defaultnya adalah none.
  • javaScriptIndexFnName: Nama fungsi JavaScript UDF yang menentukan metadata _index untuk disertakan dengan dokumen dalam permintaan massal. Setelan defaultnya adalah none.
  • propertyAsId: Properti dalam dokumen yang diindeks yang nilainya menentukan metadata _id untuk disertakan dengan dokumen dalam permintaan massal. Lebih diprioritaskan daripada UDF _id. Setelan defaultnya adalah none.
  • javaScriptIdFnGcsPath: Jalur Cloud Storage ke sumber UDF JavaScript untuk fungsi yang menentukan metadata _id yang akan disertakan dengan dokumen dalam permintaan massal. Setelan defaultnya adalah none.
  • javaScriptIdFnName: Nama fungsi JavaScript UDF yang menentukan metadata _id untuk disertakan dengan dokumen dalam permintaan massal. Setelan defaultnya adalah none.
  • javaScriptTypeFnGcsPath: Jalur Cloud Storage ke sumber UDF JavaScript untuk fungsi yang menentukan metadata _type yang akan disertakan dengan dokumen dalam permintaan massal. Setelan defaultnya adalah none.
  • javaScriptTypeFnName: Nama fungsi JavaScript UDF yang menentukan metadata _type untuk disertakan dengan dokumen dalam permintaan massal. Setelan 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. Setelan defaultnya adalah none.
  • javaScriptIsDeleteFnName: Nama fungsi JavaScript UDF yang menentukan apakah dokumen akan dihapus, bukan disisipkan atau diperbarui. Fungsi ini menampilkan nilai string true atau false. Setelan defaultnya adalah none.
  • usePartialUpdate: Apakah akan menggunakan update parsial (update, bukan membuat atau mengindeks, yang memungkinkan dokumen parsial) dengan permintaan Elasticsearch. Nilai defaultnya adalah false.
  • bulkInsertMethod: Apakah akan menggunakan INDEX (mengindeks, memungkinkan operasi upsert) atau CREATE (membuat, error pada _id duplikat) dengan permintaan massal Elasticsearch. Setelan defaultnya adalah CREATE.
  • trustSelfSignedCerts: Apakah akan memercayai sertifikat yang ditandatangani sendiri atau tidak. Instance Elasticsearch yang diinstal mungkin memiliki sertifikat yang ditandatangani sendiri. Setel ini ke benar (true) untuk melewati validasi pada sertifikat SSL. (Default: 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. Setelan defaultnya adalah false.
  • apiKeyKMSEncryptionKey: Kunci Cloud KMS untuk mendekripsi kunci API. Parameter ini wajib diisi jika apiKeySource ditetapkan ke KMS. Jika parameter ini diberikan, teruskan string apiKey terenkripsi. 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 disetel 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 terenkripsi 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 disetel, akan menggantikan waktu tunggu percobaan ulang maksimum default dan waktu tunggu soket default (30000 md) 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 diubah menjadi string untuk dimasukkan ke dalam 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 atau tidak. 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 meng-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 the Pub/Sub to Elasticsearch template.
  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