E-MapReduce:FAQ Impor Data

Apakah Stream Load dapat mengenali nama kolom di baris pertama?

Tidak. Stream Load memperlakukan baris pertama sebagai data biasa — tidak dapat mengenali atau melewati header kolom.

Jika file sumber Anda memiliki nama kolom di baris pertama, gunakan salah satu pendekatan berikut:

• Modifikasi tool ekspor agar mengekspor tanpa header kolom.

• Hapus baris pertama sebelum mengimpor: sed -i '1d' filename

• Filter baris header dengan menambahkan -H "where: column_name != 'Column Name'" ke pernyataan Stream Load. StarRocks mengonversi string header ke tipe data target dan menyaringnya. Jika konversi gagal, null dikembalikan — pastikan kolom yang terpengaruh tidak memiliki kendala NOT NULL.

• Izinkan satu baris error dengan menambahkan -H "max_filter_ratio:0.01". Ini memungkinkan baris pertama gagal tanpa membatalkan pekerjaan. ErrorURL dalam respons tetap akan menandai error tersebut, tetapi impor berhasil. Pertahankan nilai ini rendah untuk menghindari menyembunyikan error data nyata.

Apakah saya perlu mengonversi tipe data kolom kunci partisi selama Stream Load?

Ya, tetapi StarRocks menangani konversinya. Definisikan tipe kolom target dalam skema tabel dan gunakan ekspresi kolom dalam pernyataan Stream Load.

Misalnya, jika file CSV Anda memiliki kolom DATE dengan nilai dalam format 202106.00 dan tabel mengharapkan tipe DATE, petakan seperti ini:

-H "columns: NO,DATE_1, VERSION, PRICE, DATE=LEFT(DATE_1,6)"

DATE_1 adalah placeholder sementara untuk nilai mentah. LEFT(DATE_1,6) mengekstraksi enam karakter pertama dan menetapkan hasilnya ke DATE. Cantumkan semua kolom sumber berdasarkan nama sementaranya sebelum memanggil fungsi konversi apa pun. Fungsi konversi harus berupa fungsi skalar (non-agregat dan non-window).

"body exceed max size: 10737418240, limit: 10737418240"

Penyebab: File sumber melebihi batas Stream Load sebesar 10 GB.

Perbaikan:

• Pisahkan file menjadi bagian-bagian lebih kecil: seq -w 0 n

• Atau tingkatkan batas dengan memperbarui streaming_load_max_mb:

  curl -XPOST http://<be_host>:<http_port>/api/update_config?streaming_load_max_mb=<file_size>

  Lihat Konfigurasi parameter untuk parameter BE lainnya.

Bagaimana cara meningkatkan performa impor Routine Load?

Metode 1: Tingkatkan paralelisme tugas

Ini meningkatkan penggunaan CPU dan dapat menghasilkan lebih banyak versi data.

Paralelisme tugas dibatasi oleh:

min(alive_be_number, partition_number, desired_concurrent_number, max_routine_load_task_concurrent_num)

Jika alive_be_number dan partition_number keduanya melebihi desired_concurrent_number dan max_routine_load_task_concurrent_num, tingkatkan keduanya untuk meningkatkan paralelisme.

Contoh: 7 partisi, 5 node BE aktif, max_routine_load_task_concurrent_num = 5. Ubah desired_concurrent_number dari 3 menjadi 5. Paralelisme efektif: min(5, 7, 5, 5) = 5.

• Untuk pekerjaan baru: atur desired_concurrent_number dalam CREATE ROUTINE LOAD.

• Untuk pekerjaan yang sudah ada: perbarui dengan ALTER ROUTINE LOAD.

Untuk max_routine_load_task_concurrent_num, lihat Konfigurasi parameter.

Metode 2: Tingkatkan data yang dikonsumsi per tugas

Ini meningkatkan latensi impor.

Setiap tugas Routine Load mengonsumsi data hingga mencapai batas yang ditetapkan oleh max_routine_load_batch_size (byte) atau routine_load_task_consume_second (detik). Untuk mendiagnosis batas mana yang mengikat, periksa be/log/be.INFO:

I0325 20:27:50.410579 15259 data_consumer_group.cpp:131] consumer group done: 41448fb1a0ca59ad-30e34dabfa7e47a0. consume time(ms)=3261, received rows=179190, received bytes=9855450, eos: 1, left_time: -261, left_bytes: 514432550, blocking get time(us): 3065086, blocking put time(us): 24855

• Jika left_bytes >= 0: batas waktu (routine_load_task_consume_second) yang mengikat. Tingkatkan routine_load_task_consume_second.

• Jika left_bytes < 0: batas byte (max_routine_load_batch_size) yang mengikat. Tingkatkan max_routine_load_batch_size.

Lihat Konfigurasi parameter untuk detail selengkapnya.

Setelah menjalankan SHOW ROUTINE LOAD, status pekerjaan adalah PAUSED atau CANCELLED. Apa yang harus saya lakukan?

Status: PAUSED — "Broker: Offset out of range"

Penyebab: Offset konsumen tidak lagi ada di partisi Kafka.

Perbaikan: Jalankan SHOW ROUTINE LOAD dan periksa bidang Progress untuk offset terbaru. Jika pesan pada offset tersebut telah hilang, maka:

• Offset diatur ke titik masa depan saat pekerjaan dibuat.

• Kafka membersihkan pesan sebelum dikonsumsi. Sesuaikan log.retention.hours dan log.retention.bytes berdasarkan kecepatan impor Anda.

Status: PAUSED — jumlah baris error melebihi batas

Penyebab: Jumlah baris error melebihi max_error_number.

Perbaikan: Periksa ReasonOfStateChanged dan ErrorLogUrls:

• Jika format data dapat diperbaiki, perbaiki lalu jalankan RESUME ROUTINE LOAD.

• Jika format data tidak dapat diurai oleh StarRocks, tingkatkan max_error_number:

  1. Jalankan SHOW ROUTINE LOAD untuk memeriksa nilai saat ini.

  2. Jalankan ALTER ROUTINE LOAD untuk menaikkannya.

  3. Jalankan RESUME ROUTINE LOAD.

Status: CANCELLED

Penyebab: Terjadi pengecualian yang tidak dapat dipulihkan (misalnya, tabel target dihapus).

Perbaikan: Periksa ReasonOfStateChanged dan ErrorLogUrls. Pekerjaan dengan status CANCELLED tidak dapat dilanjutkan — buat pekerjaan baru.

Apakah Routine Load menjamin semantik tepat-sekali saat mengimpor dari Kafka?

Ya. Setiap tugas impor dijalankan sebagai transaksi terpisah. Jika transaksi gagal, node FE tidak memperbarui offset konsumen partisi. Saat node FE menjadwalkan ulang tugas, tugas dimulai dari offset terakhir yang disimpan, sehingga menjamin pengiriman tepat-sekali.

"Broker: Offset out of range"

Jalankan SHOW ROUTINE LOAD untuk mendapatkan offset terbaru dan verifikasi bahwa pesan pada offset tersebut ada di Kafka. Penyebab umum:

• Offset masa depan ditentukan saat pekerjaan dibuat.

• Kafka menghapus pesan sebelum pekerjaan impor mengonsumsinya. Sesuaikan log.retention.hours dan log.retention.bytes agar sesuai dengan kecepatan impor Anda.

Apakah saya dapat menjalankan ulang pekerjaan Broker Load yang berstatus FINISHED?

Tidak. Label pekerjaan yang telah selesai tidak dapat digunakan kembali, hal ini mencegah impor duplikat. Untuk mengimpor kembali data yang sama, jalankan SHOW LOAD untuk melihat konfigurasi pekerjaan asli dan buat pekerjaan baru dengan label baru.

Bidang tanggal 8 jam lebih lambat dari yang diharapkan setelah mengimpor data HDFS

Penyebab: Tabel atau pekerjaan impor dibuat dengan zona waktu UTC+8, tetapi server berjalan di UTC+0. StarRocks menambahkan 8 jam ke nilai yang disimpan.

Perbaikan: Hapus parameter timezone saat membuat tabel StarRocks.

"ErrorMsg: type:ETL_RUN_FAIL; msg:Cannot cast '\<slot 6\>' from VARCHAR to ARRAY\<VARCHAR(30)\>"

Cause: Nama kolom dalam file ORC tidak sesuai dengan nama kolom dalam tabel StarRocks. Ketika nama berbeda, StarRocks memanggil pernyataan SET dan memanggil fungsi cast() untuk inferensi tipe, yang dapat gagal.

Perbaikan: Pastikan nama kolom dalam file ORC sesuai persis dengan nama kolom dalam tabel target. Hal ini memungkinkan StarRocks melewati pernyataan SET dan fungsi cast.

Tidak ada error saat membuat pekerjaan Broker Load, tetapi tidak ada data yang muncul

Broker Load bersifat asinkron. Pernyataan CREATE LOAD yang berhasil hanya berarti pekerjaan telah diajukan, bukan telah selesai. Jalankan SHOW LOAD untuk memeriksa status pekerjaan dan pesan error, lalu sesuaikan parameter dan ajukan ulang jika diperlukan.

"failed to send batch" atau "TabletWriter add batch with unknown id"

Penyebab: Terjadi timeout penulisan data.

Perbaikan: Tingkatkan variabel sistem query_timeout dan parameter BE streaming_load_rpc_max_alive_time_sec. Lihat Konfigurasi parameter.

"LOAD-RUN-FAIL; msg:OrcScannerAdapter::init_include_columns. col name = xxx not found"

Saat mengimpor file Parquet atau ORC, nama kolom dalam header file harus sesuai dengan nama dalam tabel StarRocks. Jika berbeda, gunakan pemetaan kolom dalam klausa SET:

(tmp_c1,tmp_c2)
SET
(
   id=tmp_c2,
   name=tmp_c1
)

Jika file ORC dihasilkan oleh versi Hive lama dengan header (_col0, _col1, _col2, ...), konfigurasikan pemetaan kolom secara eksplisit melalui SET.

Pekerjaan impor macet dan tidak selesai

Dalam log fe.log node FE, cari ID pekerjaan impor menggunakan label pekerjaan. Lalu cari ID pekerjaan tersebut dalam log be.INFO node BE untuk mengidentifikasi penyebabnya.

Bagaimana cara mengakses kluster HDFS berkeandalan tinggi?

Konfigurasikan parameter berikut untuk mengaktifkan failover otomatis NameNode:

Contoh konfigurasi menggunakan autentikasi simple:

(
    "username"="user",
    "password"="passwd",
    "dfs.nameservices" = "my-ha",
    "dfs.ha.namenodes.my-ha" = "my_namenode1,my_namenode2",
    "dfs.namenode.rpc-address.my-ha.my-namenode1" = "nn1-host:rpc_port",
    "dfs.namenode.rpc-address.my-ha.my-namenode2" = "nn2-host:rpc_port",
    "dfs.client.failover.proxy.provider" = "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"
)

Anda juga dapat menulis pengaturan ini ke hdfs-site.xml. Jika menggunakan proses broker untuk mengakses HDFS, cukup tentukan path file dan informasi autentikasi.

Bagaimana cara mengonfigurasi ViewFs dalam Federasi HDFS?

Salin core-site.xml dan hdfs-site.xml dari konfigurasi ViewFs ke direktori broker/conf.

Jika pengaturan mencakup sistem file kustom, salin file .jar terkait ke direktori broker/lib.

"Can't get Kerberos realm" saat mengakses kluster EMR yang diaktifkan Kerberos

1. Verifikasi bahwa /etc/krb5.conf ada di setiap mesin broker.

2. Jika error tetap muncul, tambahkan -Djava.security.krb5.conf:/etc/krb5.conf ke variabel JAVA_OPTS dalam skrip startup broker.

Mengapa memasukkan satu baris dengan INSERT INTO memakan waktu 50–100 ms?

INSERT INTO dirancang untuk impor batch. Waktu untuk memasukkan satu baris sama dengan waktu untuk batch karena operasi selalu membuka transaksi penuh. Hindari menggunakan INSERT INTO untuk memuat baris individual dalam beban kerja OLAP (online analytical processing).

"index channel has intolerable failure" saat menjalankan INSERT INTO SELECT

Penyebab: Timeout RPC Stream Load.

Perbaikan: Perbarui parameter berikut dalam be.conf dan restart kluster:

"execute timeout" saat menjalankan INSERT INTO SELECT pada dataset besar

Penyebab: Kueri melebihi timeout sesi.

Perbaikan: Tingkatkan query_timeout untuk sesi (default: 600 detik):

SET query_timeout = <value>;

"Could not execute SQL statement. Reason: ValidationException: One or more required options are missing" saat menjalankan pekerjaan Flink

Penyebab: Beberapa aturan (misalnya, [table-rule.1] dan [table-rule.2]) didefinisikan dalam file config_prod.conf StarRocks-migrate-tools (SMT), tetapi bidang wajib tidak lengkap untuk satu atau beberapa aturan.

Perbaikan: Periksa bahwa setiap aturan memiliki database, tabel, dan konektor Flink yang dikonfigurasi.

Bagaimana Flink merestart tugas yang gagal?

Flink menggunakan mekanisme checkpointing yang dikombinasikan dengan kebijakan restart. Untuk mengaktifkan checkpointing dengan kebijakan restart fixed-delay, tambahkan berikut ke flink-conf.yaml:

# Interval checkpoint dalam milidetik
execution.checkpointing.interval: 300000
state.backend: filesystem
state.checkpoints.dir: file:///tmp/flink-checkpoints-directory

Bagaimana cara menghentikan pekerjaan Flink dan memulihkannya ke status sebelumnya?

Gunakan titik simpan (savepoint). Titik simpan adalah snapshot konsisten dari status eksekusi pekerjaan streaming, yang dipicu secara manual. Lihat Savepoints.

Hentikan pekerjaan dan buat titik simpan:

bin/flink stop --type [native/canonical] --savepointPath [:targetDirectory] :jobId

Untuk memulihkan pekerjaan, tentukan titik simpan saat mengajukan ulang:

./flink run -c com.starrocks.connector.flink.tools.ExecuteSQL -s savepoints_dir/savepoints-xxxxxxxx flink-connector-starrocks-xxxx.jar -f flink-create.all.sql

Impor gagal saat menggunakan semantik tepat-sekali

Jika Anda melihat error seperti:

{
    "Status": "Fail",
    "Message": "timeout by txn manager",
    "WriteDataTimeMs": 120278,
    ...
}

Penyebab: Nilai sink.properties.timeout lebih kecil daripada interval checkpoint Flink.

Perbaikan: Tingkatkan sink.properties.timeout sehingga melebihi interval checkpoint.

Setelah menggunakan flink-connector-jdbc_2.11, timestamp di StarRocks 8 jam lebih awal daripada di Flink

Atur server-time-zone ke Asia/Shanghai pada tabel sink Flink dan tambahkan &serverTimezone=Asia/Shanghai ke parameter url:

CREATE TABLE sk (
    sid int,
    local_dtm TIMESTAMP,
    curr_dtm TIMESTAMP
)
WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://192.168.**.**:9030/sys_device?characterEncoding=utf-8&serverTimezone=Asia/Shanghai',
    'table-name' = 'sink',
    'driver' = 'com.mysql.jdbc.Driver',
    'username' = 'sr',
    'password' = 'sr123',
    'server-time-zone' = 'Asia/Shanghai'
);

Kafka pada kluster StarRocks berhasil diimpor, tetapi Kafka pada mesin lain gagal

Jika Anda melihat:

failed to query wartermark offset, err: Local: Bad message format

Penyebab: StarRocks tidak dapat menyelesaikan hostname Kafka.

Perbaikan: Tambahkan hostname Kafka ke /etc/hosts di setiap node kluster StarRocks.

Memori node BE terisi penuh dan CPU mencapai 100% meskipun tidak ada kueri yang berjalan

Node BE mengumpulkan statistik secara berkala dan mengelola memori berdasarkan konfigurasinya. Secara default, StarRocks mengembalikan memori ke sistem operasi (OS) hanya setelah memori yang digunakan melebihi tc_use_memory_min (default: 10737418240 byte, sekitar 10 GiB).

Untuk mengubah ambang batas ini, perbarui tc_use_memory_min dalam konfigurasi BE. Di Konsol EMR, buka tab Configure pada halaman layanan StarRocks dan pilih tab be.conf. Lihat Konfigurasi parameter.

Mengapa node BE menunda pengembalian memori ke OS?

Alokasi memori mahal. Saat StarRocks meminta sejumlah besar memori dari OS, sistem menyisakan kapasitas tambahan untuk penggunaan di masa depan. Memori yang tidak digunakan dikembalikan ke OS secara bertahap, bukan segera. Untuk memvalidasi perilaku ini, pantau penggunaan memori dalam periode panjang di lingkungan pengujian Anda untuk memastikan memori akhirnya dilepaskan.

Sistem tidak dapat mengurai dependensi konektor Flink

Penyebab: File /etc/maven/settings.xml tidak dikonfigurasi untuk menggunakan repositori publik Alibaba Cloud, sehingga beberapa dependensi tidak dapat diselesaikan.

Perbaikan: Atur alamat repositori publik Alibaba Cloud ke https://maven.aliyun.com/repository/public dalam /etc/maven/settings.xml.

Apakah sink.buffer-flush.interval-ms tetap berlaku saat interval checkpoint diatur?

Ya. Operasi flush tidak dibatasi oleh interval checkpoint. Flush dipicu ketika salah satu ambang batas berikut tercapai:

sink.buffer-flush.max-rows
sink.buffer-flush.max-bytes
sink.buffer-flush.interval-ms

Interval checkpoint mengontrol semantik tepat-sekali. Parameter sink.buffer-flush.interval-ms mengontrol perilaku flush setidaknya-sekali. Keduanya beroperasi secara independen.

Apakah saya dapat memperbarui data yang diimpor oleh DataX?

Ya. StarRocks versi terbaru mendukung pembaruan data dalam tabel yang dibuat dengan model primary key menggunakan DataX. Tambahkan bidang _op di bagian reader file konfigurasi JSON.

Bagaimana cara menangani kata kunci DataX untuk menghindari error impor?

Apit kata kunci dengan backtick (` ``).

"When running with master 'yarn' either HADOOP-CONF-DIR or YARN-CONF-DIR must be set in the environment"

Konfigurasikan variabel lingkungan HADOOP-CONF-DIR dalam skrip spark-env.sh klien Spark.

"Cannot run program 'xxx/bin/spark-submit': error=2, No such file or directory"

Penyebab: Parameter spark_home_default_dir tidak ada atau mengarah ke direktori yang salah.

Perbaikan: Atur ke direktori root klien Spark yang benar.

"File xxx/jars/spark-2x.zip does not exist"

Penyebab: Parameter spark-resource-path tidak mengarah ke file ZIP yang telah dipaketkan.

Perbaikan: Verifikasi bahwa path sesuai dengan nama file aktual.

"yarn client does not exist in path: xxx/yarn-client/hadoop/bin/yarn"

Penyebab: Tidak ada file yang dapat dieksekusi yang dikonfigurasi untuk yarn-client-path.

Perbaikan: Atur ke path biner klien YARN yang benar.