http://starrocks.io logo
Join Slack
Powered by
<@U0A71G31CDV> How StarRocks v 4.1.1 ingest data i...
# questions-and-troubleshooting
d
@Rocky How StarRocks v 4.1.1 ingest data in iceberg? Does it wait for filesize to be a specifc size and the push it toiceberg? HOw can I control filesize on iceberg on s3?
r
In StarRocks v4.1.1, data ingestion into Iceberg (typically via 
INSERT INTO
) is handled by a distributed sink mechanism that buffers data in memory before writing it to storage (like S3). Does it wait for a specific file size? StarRocks does not wait indefinitely for a file to reach a specific size, but it uses target-based buffering. It attempts to accumulate enough data to reach a target file size to avoid the "small file problem." * Buffering & Spilling: Starting from v4.0, StarRocks introduced 
enable_connector_sink_spill
(enabled by default). This feature allows the sink to spill data to disk if memory is insufficient, which helps maintain larger file sizes even when writing to many partitions simultaneously. * Best Effort: The system applies the file size limit on a "best-effort" basis. A file will be finalized and pushed to S3 when it reaches the target size, or when the ingestion task completes. How to control file size on Iceberg (S3) You can control the size of the files generated during ingestion using the following session variables: 1. 
connector_sink_target_max_file_size
 * Description: Specifies the target maximum size for data files written to Iceberg. * Default: 
1073741824
(1 GB). * Usage: 
SET connector_sink_target_max_file_size = 536870912; -- Set to 512MB
2. 
enable_connector_sink_spill
(BE Configuration) * Description: When enabled, StarRocks buffers data and spills to local disk if needed to ensure that it can produce larger files rather than many small ones when memory is tight. * Default: 
true
. 3. 
pipeline_sink_dop
 * Description: Controls the parallelism (degree of parallelism) of the sink. * Effect on File Size: If this is set too high, the data is split across more concurrent writers, which can result in smaller files for each writer. Reducing this can help consolidate data into fewer, larger files. 4. Iceberg Table Properties * StarRocks also respects Iceberg's native properties if set on the table, such as 
write.parquet.compression-codec
. For compression specifically, you can also use the session variable 
connector_sink_compression_codec
. New in v4.1.1 (and v4.1.0) * 
iceberg_commit_queue_max_size
: This FE configuration (introduced in v4.1.0) helps manage high-concurrency writes by queuing commit operations. While it doesn't directly change the file size, it improves the stability of the ingestion process when many tasks are pushing files to the same Iceberg table simultaneously. References • sql-reference/System_variable.md • administration/management/BE_configuration.md • administration/management/FE_configuration.md • data_source/catalog/iceberg/iceberg_catalog.md
d
what does it mean by ingestion task completes? @Rocky Like I have data coming in from upstream which I want ingestion to continue and data comes around 10 to 15 mins in parquet files so does that mean it'll take 1 file as ingestion process?
r
In StarRocks v4.1.1, the phrase "ingestion task completes" refers to the end of a single SQL execution (typically an 
INSERT INTO ... SELECT ...
statement). Because Iceberg follows a transactional model, every 
INSERT
statement must commit its changes to the table metadata to make the data visible. When that SQL statement finishes, StarRocks is forced to finalize and push all currently buffered data to S3, regardless of whether it has reached the target file size. Why you are seeing small files In your scenario where parquet files arrive every 10–15 minutes: * If you run one 
INSERT INTO
for every single file that arrives, each load is a separate "ingestion task." * StarRocks cannot "hold" a file open across different SQL statements to wait for more data. * Therefore, if an incoming batch is only 50MB but your 
connector_sink_target_max_file_size
is 1GB, StarRocks will still write a 50MB file because the transaction must close when the SQL finishes. How to control file size on S3 To manage file sizes and prevent the "small file problem" in your Iceberg catalog on S3, you can use the following strategies: 1. Session Variables (Within a single task) If a single 
INSERT
statement is producing many small files, use these variables to consolidate them: * 
connector_sink_target_max_file_size
: Sets the goal for file size (default is 1GB). * 
pipeline_sink_dop
: Lower this value (e.g., to 
1
) if you want to reduce the number of parallel writers. High parallelism splits data into more files. * 
enable_connector_sink_spill
: Keep this 
true
(default). It allows StarRocks to buffer more data in memory/disk before writing to storage. 2. Change the Ingestion Strategy Instead of one 
INSERT
per file, you can: * Batch upstream files: Wait until you have several parquet files (e.g., 1 hour's worth) and run one 
INSERT INTO iceberg_table SELECT * FROM FILES(...)
for all of them at once. * Use a Materialized View: Create an Asynchronous Materialized View (MV) on top of your source files. You can configure the MV to refresh on a schedule that allows more data to accumulate before writing to the target Iceberg table. 3. Post-Ingestion Maintenance (Compaction) StarRocks provides a specific procedure to clean up small files after they have been written to Iceberg. You can run this periodically (e.g., once a day):
Copy code
sql
-- Merges small files in 'my_table' that are smaller than 128MB
ALTER TABLE my_catalog.my_db.my_table 
EXECUTE rewrite_data_files("min_file_size_bytes"=134217728);
Summary StarRocks will not wait for future data to arrive to fill a file; it pushes whatever is in the current task. To get larger files, you must either batch more data into a single 
INSERT
command or use the 
rewrite_data_files
 procedure to merge them later. References • knowledge/usage/best-practice.md • data_source/catalog/iceberg/iceberg_catalog.md • best_practices/primarykey_table.md • unloading/unload_using_insert_into_files.md • faq/shared_data_faq.md
d
So we can ingest more tahn 1 parquest file in 1 go? is there something lke manifest file system in starrocks? like redshift able to ingest multiple files in 1 go using manifest file which containes name of many files? @Rocky
r
Yes, you can ingest multiple Parquet files in a single operation. While StarRocks does not have a direct equivalent to Redshift’s JSON manifest file (where you list individual URIs), it provides several ways to achieve the same goal. 1. Ingesting Multiple Files with 
FILES()
 The most common way to load multiple files in one go is to use the 
FILES()
table function with wildcards or recursive listing. * Wildcards: You can use 
*
or 
?
to match multiple files.
Copy code
sql
    INSERT INTO my_iceberg_table
    SELECT * FROM FILES(
        "path" = "<s3://my-bucket/incoming/2023-10-*.parquet>",
        "format" = "parquet",
        -- ... credentials ...
    );
* Recursive Listing: If your files are nested in subfolders, you can set 
list_recursively
to 
true
.
Copy code
sql
    INSERT INTO my_iceberg_table
    SELECT * FROM FILES(
        "path" = "<s3://my-bucket/incoming/>",
        "format" = "parquet",
        "list_recursively" = "true",
        -- ... credentials ...
    );
2. Loading Specific Files (Broker Load) If you need to load a specific list of files that don't follow a simple pattern (like a Redshift manifest would), you can use Broker Load. This allows you to specify multiple 
DATA INFILE
clauses in a single asynchronous job.
Copy code
sql
LOAD LABEL my_load_label
(
    DATA INFILE("<s3://my-bucket/file_a.parquet>") INTO TABLE my_iceberg_table FORMAT AS "parquet",
    DATA INFILE("<s3://my-bucket/specific_folder/file_b.parquet>") INTO TABLE my_iceberg_table FORMAT AS "parquet"
)
WITH BROKER ("aws.s3.access_key" = "...", "aws.s3.secret_key" = "...");
3. Using 
Pipe
for Continuous Ingestion If files are arriving every 10–15 minutes, you can use the Pipe feature (introduced in v3.2). A Pipe maintains a file queue for a table. As new files arrive in S3, the Pipe detects them and ingests them. * Advantage: It handles the batching and retry logic for you. * Control: It allows you to ingest files as they appear without manually triggering an 
INSERT
for every single file. Why this helps with your File Size issue In your previous question, you were concerned about small files. By using wildcards or Broker Load to group many files into a single ingestion task, StarRocks will: 1. Open a single transaction for the whole batch. 2. Buffer the data from all those files together. 3. Write out larger files to Iceberg (targetting your 
connector_sink_target_max_file_size
), rather than creating one small Iceberg file per source file. Summary of Differences Feature | StarRocks 
FILES()
 | Redshift Manifest Patterns | Supports 
*
, 
?
, 
[]
wildcards. | Limited; usually requires listing every file. Listing | Can list directories recursively. | Requires an external JSON file listing files. Multi-Path | Use wildcards or multiple 
DATA INFILE
in Broker Load. | Points to one JSON file containing all paths. Recommendation for your 10-15 min Parquet files: Instead of running an 
INSERT
every 10 minutes, set up a StarRocks Pipe on the S3 bucket or run a scheduled 
INSERT INTO ... SELECT * FROM FILES(...)
with a wildcard that picks up all "new" files every hour. This will significantly improve your file size distribution on S3. References • loading/s3.md • loading/Loading_data_template.md • loading/BrokerLoad.md
d
will we abel to do deduplication in iceberg ingestion where read the data from s3 and deduplciate it and put itn s3 all at once? @Rocky
r
In StarRocks v4.1.1, you can perform deduplication during Iceberg ingestion in a single operation. You have two primary ways to do this depending on whether you are deduplicating data within the incoming files or against existing data in the Iceberg table. 1. Deduplicating "Within the Batch" (Incoming Files) If your source files on S3 contain duplicate records and you want to write only unique records to Iceberg, you can use standard SQL deduplication logic inside an 
INSERT INTO
statement. Since StarRocks processes the 
SELECT
query in its vectorized engine before pushing it to the Iceberg sink, this happens "all at once" in one transaction. * Using 
DISTINCT
or `GROUP BY`:
Copy code
sql
    INSERT INTO my_iceberg_catalog.my_db.my_table
    SELECT DISTINCT *
    FROM FILES(
        "path" = "<s3://my-bucket/incoming/*.parquet>",
        "format" = "parquet",
        ...
    );
* Using Window Functions (to keep the "latest" record): If you have multiple versions of the same ID and want the one with the latest timestamp:
Copy code
sql
    INSERT INTO my_iceberg_catalog.my_db.my_table
    SELECT id, name, update_time
    FROM (
        SELECT *,
               ROW_NUMBER() OVER(PARTITION BY id ORDER BY update_time DESC) as rn
        FROM FILES(...)
    ) t
    WHERE rn = 1;
2. Deduplicating "Against Existing Data" (Upsert/Merge) If you need to check if the data already exists in the target Iceberg table and either update it or skip it (to prevent duplicates across different ingestion runs), StarRocks supports 
MERGE INTO
 for Iceberg (introduced in newer versions like v3.3+ and refined in v4.x).
Copy code
sql
MERGE INTO my_iceberg_catalog.my_db.target_table AS t
USING (SELECT * FROM FILES(...)) AS s
ON t.id = s.id
WHEN MATCHED THEN 
    UPDATE SET t.name = s.name, t.update_time = s.update_time
WHEN NOT MATCHED THEN 
    INSERT (id, name, update_time) VALUES (s.id, s.name, s.update_time);
Note: Iceberg 
MERGE INTO
support requires the table to be an Iceberg v2 table with support for row-level deletes/updates. Key Technical Details * Vectorized Execution: StarRocks performs the deduplication (the 
GROUP BY
or 
ROW_NUMBER
logic) in its distributed compute layer (BE nodes). Only the final, deduplicated result set is then formatted into Parquet and pushed to S3. * Transactional Safety: Because Iceberg is a transactional table format, the entire 
INSERT
or 
MERGE
operation is atomic. If the deduplication query fails halfway through, no partial data is committed to the Iceberg table. * Memory Management: For very large deduplication tasks, StarRocks uses its pipeline engine to spill to disk if the intermediate deduplication state (e.g., the 
DISTINCT
set) exceeds available RAM. Summary To ingest and deduplicate all at once: 1. Read from S3 using the 
FILES()
function. 2. Deduplicate in-memory using 
DISTINCT
, 
GROUP BY
, or 
ROW_NUMBER()
. 3. Sink to S3 (Iceberg) via 
INSERT INTO
or 
MERGE INTO
. References • data_source/catalog/iceberg/DML.md • sql-reference/sql-statements/loading_unloading/INSERT.md • loading/InsertInto.md • data_source/catalog/iceberg/iceberg_catalog.md References • release_notes/release-3.3.md • data_source/catalog/iceberg/DML.md
Open in Slack
PreviousNext
Powered by