E-MapReduce:Routine Load

Prasyarat

• Anda dapat terhubung ke kluster Kafka yang tidak menggunakan otentikasi maupun yang menggunakan otentikasi SSL.

• Format pesan berikut didukung:

  • Format CSV, di mana setiap pesan merupakan satu baris tanpa karakter newline di akhir.

  • Format JSON.

• Tipe data Array tidak didukung.

• Hanya Kafka versi 0.10.0.0 ke atas yang didukung.

Buat job impor

• Sintaks

  CREATE ROUTINE LOAD <database>.<job_name> ON <table_name>
      [COLUMNS TERMINATED BY "column_separator" ,]
      [COLUMNS (col1, col2, ...) ,]
      [WHERE where_condition ,]
      [PARTITION (part1, part2, ...)]
      [PROPERTIES ("key" = "value", ...)]
      FROM [DATA_SOURCE]
      [(data_source_properties1 = 'value1',
      data_source_properties2 = 'value2',
      ...)]

  Tabel berikut menjelaskan parameter-parameter tersebut.

  Parameter	Wajib	Deskripsi
  job_name	Ya	Nama job impor. Nama database impor dapat ditempatkan di awal. Nama biasanya dalam format timestamp ditambah nama tabel. Nama job harus unik dalam satu database.
  table_name	Ya	Nama tabel tujuan.
  COLUMNS TERMINATED clause	Tidak	Pemisah kolom dalam file data sumber. Nilai default: \t.
  COLUMNS clause	Tidak	Pemetaan antara kolom dalam file data sumber dan kolom dalam tabel tujuan. Kolom yang dipetakan: Misalnya, tabel tujuan memiliki tiga kolom, col1, col2, dan col3, sedangkan file data sumber memiliki empat kolom, dan kolom pertama, kedua, dan keempat dalam tabel tujuan masing-masing sesuai dengan col2, col1, dan col3 dalam file data sumber. Dalam kasus ini, klausa dapat ditulis sebagai COLUMNS (col2, col1, temp, col3). Kolom temp tidak ada dan digunakan untuk melewati kolom ketiga dalam file data sumber. Kolom turunan: StarRocks tidak hanya dapat membaca data dalam kolom file data sumber tetapi juga menyediakan operasi pemrosesan pada kolom data. Misalnya, kolom col4 ditambahkan ke tabel tujuan, dan nilai col4 sama dengan nilai col1 ditambah nilai col2. Dalam kasus ini, klausa dapat ditulis sebagai COLUMNS (col2, col1, temp, col3, col4 = col1 + col2).
  WHERE clause	Tidak	Kondisi filter yang ingin Anda gunakan untuk menyaring baris yang tidak diperlukan. Kondisi filter dapat ditentukan pada kolom yang dipetakan atau kolom turunan. Misalnya, jika hanya baris dengan k1 lebih besar dari 100 dan k2 sama dengan 1000 yang diimpor, klausa dapat ditulis sebagai WHERE k1 > 100 and k2 = 1000.
  PARTITION clause	Tidak	Partisi tabel tujuan. Jika Anda tidak menentukan partisi, file data sumber akan secara otomatis diimpor ke partisi yang sesuai.
  PROPERTIES clause	Tidak	Parameter umum untuk job impor.
  desired_concurrent_number	Tidak	Jumlah maksimum task yang dapat dibagi dari job impor. Nilainya harus lebih besar dari 0. Nilai default: 3.
  max_batch_interval	Tidak	Durasi eksekusi maksimum setiap task. Nilai valid: 5 hingga 60. Satuan: detik. Nilai default: 10. Pada V1.15 dan versi selanjutnya, parameter ini menentukan waktu penjadwalan task. Anda dapat menentukan seberapa sering task dieksekusi. routine_load_task_consume_second dalam fe.conf menentukan jumlah waktu yang diperlukan task untuk mengonsumsi data. Nilai default: 3 detik. routine_load_task_timeout_second dalam fe.conf menentukan periode timeout eksekusi task. Nilai default: 15 detik.
  max_batch_rows	Tidak	Jumlah maksimum baris yang dapat dibaca setiap task. Nilainya harus lebih besar dari atau sama dengan 200000. Nilai default: 200000. Pada V1.15 dan versi selanjutnya, parameter ini hanya digunakan untuk menentukan rentang jendela deteksi error. Rentang jendela tersebut adalah 10 × max-batch-rows.
  max_batch_size	Tidak	Jumlah maksimum byte yang dapat dibaca setiap task. Satuan: byte. Nilai valid: 100 MB hingga 1 GB. Nilai default: 100 MB. Pada V1.15 dan versi selanjutnya, parameter ini tidak digunakan lagi. routine_load_task_consume_second dalam fe.conf menentukan jumlah waktu yang diperlukan task untuk mengonsumsi data. Nilai default: 3 detik.
  max_error_number	Tidak	Jumlah maksimum baris error yang diizinkan dalam jendela sampling. Nilainya harus lebih besar dari atau sama dengan 0. Nilai default: 0. Tidak ada baris error yang diizinkan. Penting Baris yang disaring oleh kondisi WHERE bukan merupakan baris error.
  strict_mode	Tidak	Menentukan apakah mode ketat diaktifkan. Secara default, mode ini diaktifkan. Jika tipe kolom data mentah non-null berubah menjadi NULL setelah Anda mengaktifkan mode ketat, data tersebut akan disaring. Untuk menonaktifkan mode ketat, atur parameter ini ke false.
  timezone	Tidak	Zona waktu yang digunakan untuk job impor. Secara default, nilai parameter timezone dari session digunakan. Parameter ini memengaruhi hasil semua fungsi terkait zona waktu yang terlibat dalam impor.
  DATA_SOURCE	Ya	Jenis sumber data. Atur nilainya ke KAFKA.
  data_source_properties	Tidak	Informasi tentang sumber data. Nilainya mencakup bidang-bidang berikut: kafka_broker_list: informasi koneksi broker Kafka. Format: ip:host. Pisahkan beberapa broker dengan koma (,). kafka_topic: topik Kafka yang ingin Anda langgani. Catatan Bidang kafka_broker_list dan kafka_topic wajib diisi. kafka_partitions dan kafka_offsets: partisi Kafka yang ingin Anda langgani dan offset awal setiap partisi. property: properti terkait Kafka. Bidang ini setara dengan parameter --property dalam Kafka Shell. Anda dapat menjalankan perintah HELP ROUTINE LOAD; untuk melihat sintaks lebih rinci untuk membuat job impor.

  Catatan

  Anda dapat menjalankan perintah HELP ROUTINE LOAD; untuk melihat sintaks lebih rinci untuk membuat job impor.

• Contoh ini membuat job impor Routine Load bernama example_tbl2_ordertest. Job ini tidak menggunakan autentikasi dan terus-menerus mengonsumsi pesan dari topik ordertest2 di kluster Kafka serta mengimpornya ke tabel example_tbl. Job ini mulai mengonsumsi dari offset paling awal yang tersedia di partisi yang ditentukan.

  CREATE ROUTINE LOAD load_test.example_tbl2_ordertest ON example_tbl
  COLUMNS(commodity_id, customer_name, country, pay_time, price, pay_dt=from_unixtime(pay_time, '%Y%m%d'))
  PROPERTIES
  (
      "desired_concurrent_number"="5",
      "format" ="json",
      "jsonpaths" ="[\"$.commodity_id\",\"$.customer_name\",\"$.country\",\"$.pay_time\",\"$.price\"]"
   )
  FROM KAFKA
  (
      "kafka_broker_list" ="<kafka_broker1_ip>:<kafka_broker1_port>,<kafka_broker2_ip>:<kafka_broker2_port>",
      "kafka_topic" = "ordertest2",
      "kafka_partitions" ="0,1,2,3,4",
      "property.kafka_default_offsets" = "OFFSET_BEGINNING"
  );

• Contoh berikut menunjukkan cara mengonfigurasi koneksi ke Kafka menggunakan SSL.

  -- Tentukan SSL sebagai protokol keamanan.
  "property.security.protocol" = "ssl", 
  
   -- Tentukan lokasi sertifikat CA.
  "property.ssl.ca.location" = "FILE:ca-cert",
  
  -- Jika autentikasi klien diaktifkan di server Kafka, Anda juga harus mengatur tiga parameter berikut:
  -- Lokasi kunci publik klien.
  "property.ssl.certificate.location" = "FILE:client.pem", 
  -- Lokasi kunci privat klien.
  "property.ssl.key.location" = "FILE:client.key", 
  -- Password untuk kunci privat klien.
  "property.ssl.key.password" = "******"

  Untuk informasi lebih lanjut tentang pembuatan file, lihat CREATE FILE.

  Catatan

  Saat menggunakan CREATE FILE, gunakan endpoint HTTP Object Storage Service (OSS) sebagai url. Untuk informasi lebih lanjut, lihat Gunakan endpoint yang mendukung IPv6 untuk mengakses OSS.

Lihat status job

• Tampilkan semua job Routine Load di database load_test, termasuk job yang dihentikan atau dibatalkan. Hasilnya dapat berisi satu atau beberapa baris.

  USE load_test;
  SHOW ALL ROUTINE LOAD;

• Tampilkan job Routine Load yang sedang berjalan bernama example_tbl2_ordertest di database load_test.

  SHOW ROUTINE LOAD FOR load_test.example_tbl2_ordertest;

• Di Konsol EMR StarRocks Manager, klik Metadata Management. Klik nama database target, klik Tasks, lalu lihat status job di tab Kafka Import.

*************************** 1. row ***************************

                  Id: 14093
                Name: routine_load_wikipedia
          CreateTime: 2020-05-16 16:00:48
           PauseTime: N/A
             EndTime: N/A
              DbName: default_cluster:load_test
           TableName: routine_wiki_edit
               State: RUNNING
      DataSourceType: KAFKA
      CurrentTaskNum: 1
       JobProperties: {"partitions":"*","columnToColumnExpr":"event_time,channel,user,is_anonymous,is_minor,is_new,is_robot,is_unpatrolled,delta,added,deleted","maxBatchIntervalS":"10","whereExpr":"*","maxBatchSizeBytes":"104857600","columnSeparator":"','","maxErrorNum":"1000","currentTaskConcurrentNum":"1","maxBatchRows":"200000"}
DataSourceProperties: {"topic":"starrocks-load","currentKafkaPartitions":"0","brokerList":"localhost:9092"}
    CustomProperties: {}
           Statistic: {"receivedBytes":150821770,"errorRows":122,"committedTaskNum":12,"loadedRows":2399878,"loadRowsRate":199000,"abortedTaskNum":1,"totalRows":2400000,"unselectedRows":0,"receivedBytesRate":12523000,"taskExecuteTimeMs":12043}
            Progress: {"0":"13634667"}
ReasonOfStateChanged:
        ErrorLogUrls: http://172.26.**.**:9122/api/_load_error_log?file=__shard_53/error_log_insert_stmt_47e8a1d107ed4932-8f1ddf7b01ad2fee_47e8a1d107ed4932_8f1ddf7b01ad2fee, http://172.26.**.**:9122/api/_load_error_log?file=__shard_54/error_log_insert_stmt_e0c0c6b040c044fd-a162b16f6bad53e6_e0c0c6b040c044fd_a162b16f6bad53e6, http://172.26.**.**:9122/api/_load_error_log?file=__shard_55/error_log_insert_stmt_ce4c95f0c72440ef-a442bb300bd743c8_ce4c95f0c72440ef_a442bb300bd743c8
            OtherMsg:
1 row in set (0.00 sec)

Jeda job impor

Gunakan pernyataan PAUSE untuk menjeda job impor. Job tersebut masuk ke status PAUSED, dan impor data dihentikan sementara. Job yang dijeda tidak dihentikan dan dapat dilanjutkan kembali dengan menggunakan pernyataan RESUME.

PAUSE ROUTINE LOAD FOR <job_name>;

Setelah Anda menjeda job impor, State-nya berubah menjadi PAUSED, dan informasi dalam Statistic dan Progress berhenti diperbarui. Karena job tidak dihentikan, Anda masih dapat melihatnya dengan menjalankan pernyataan SHOW ROUTINE LOAD.

Lanjutkan job impor

Gunakan pernyataan RESUME untuk melanjutkan job yang dijeda. Job tersebut sejenak masuk ke status NEED_SCHEDULE, yang menunjukkan bahwa job sedang dijadwalkan ulang. Kemudian job kembali ke status RUNNING dan melanjutkan impor data.

RESUME ROUTINE LOAD FOR <job_name>;

Hentikan job impor

Gunakan pernyataan STOP untuk menghentikan job impor. Job tersebut masuk ke status STOPPED, impor data berhenti, dan job dihentikan. Anda tidak dapat melanjutkan kembali job yang telah dihentikan.

STOP ROUTINE LOAD FOR <job_name>;