Batch and Continuous Data Ingestion

Course README · Open Lab 5 · Open Quiz 5

Module goal

Load files into Doris in batches, check that each load did what it reports, and retry a load without loading anything twice.

Learning objectives

After completing this module, you will be able to:

1. Choose a load method for a local file, files in object storage, a message stream, a database's changes, or many small writes.
2. Map CSV positions, Parquet fields, and default values to the columns of a target table.
3. Judge a load from its response and from the rows, relationships, and amounts in the target table.
4. Retry a load batch safely with its label, and tell a batch retry from business deduplication.
5. Explain what Group Commit changes for many small loads, and when their rows become visible.
6. Distinguish the grain and amount definitions of WWI historical orders, account receipts, and the simulated new orders.

Module structure

Continuous ingestion, where data keeps arriving, is optional in this course: Kafka with Routine Load, database changes with Flink CDC, and Doris's own Streaming Jobs are in the reading Continuous Ingestion, with its own quiz and the optional Labs 5A and 5B. Nothing in Modules 6 and 7 depends on them.

5.1 Choose a load method

A warehouse usually starts with history that already exists, then receives what keeps changing. The file format alone does not decide how to load: a Parquet file on your machine, a fixed set of Parquet files in object storage, and a directory of Parquet files that keeps growing need different methods.

These are choices, not steps of one pipeline. Module 4 also showed a choice that loads nothing: querying the lake table where it is.

Lab 5 practices Stream Load on two kinds of data, ten simulated new orders in a CSV file and the ten tables of the WWI history in Parquet, and Group Commit on the same ten orders sent one at a time.

5.2 Types, columns, and defaults

A table's schema is its columns, their types, and whether they accept NULL. Choose each type from what the column is used for: DECIMAL(18, 2) for an amount with two decimal places, 18 digits in all; BIGINT for an ID; DATE for a calendar day; DATETIME for the moment an event happened. Decide what NULL means in a column, such as "payment time unknown", so that nobody reads it as zero. See Data types.

A value that converts is not yet a valid business value. The text not-a-number cannot become an amount: that is a type error. The customer ID 999999 converts to an integer, yet there may be no such customer: that needs a business check against the customer dimension, which Module 6 builds.

Tell the load which column each value fills

The simulated new orders come as a CSV file without a header, one order per line, eleven values separated by commas:

order_id,customer_id,order_amount,status,event_version,event_id,
event_time,paid_amount,refund_amount,region,data_source

The columns parameter of a load names, in the file's order, the table column that each value fills. If a file puts the customer ID before the order ID and columns does not say so, every value still converts, and the two IDs are silently swapped; spot-check a few orders after a load for this reason.

A columns entry can also compute a value, such as order_amount = amount_cents / 100.0 for a source that sends cents: 18000 becomes 180.00. That belongs to the load, because the rule is about one source's format. A rule that belongs to the table, whatever loads it, can be a generated column in the table definition instead; the files and Group Commit extension compares the two.

Defaults fill only what a load leaves out

A default is the value that a column gets when a write does not supply it. A table that only ever receives new orders can default status to CREATED; a table that receives orders in every state should take the state from the source. Facts such as a payment must come from the source, never from a default.

SQL reading example: not executed in the lab. To try it, use an orders_defaults_reading table that does not yet exist in an isolated lab database, and run it only once; do not write this example into orders_imported.

CREATE TABLE orders_defaults_reading (
    order_id BIGINT NOT NULL,
    status VARCHAR(20) NOT NULL DEFAULT "CREATED"
) DUPLICATE KEY(order_id)
DISTRIBUTED BY HASH(order_id) BUCKETS 1
PROPERTIES("replication_num"="1");
INSERT INTO orders_defaults_reading (order_id) VALUES (901001);
INSERT INTO orders_defaults_reading (order_id, status) VALUES (901002, 'PAID');
SELECT order_id, status FROM orders_defaults_reading ORDER BY order_id;

The first row leaves out status and gets (901001, CREATED); the second supplies it and keeps (901002, PAID). A default handles a missing column; it does not repair a wrong value.

5.3 Stream Load: request, response, and retries

Stream Load sends a file to Doris in one HTTP PUT request. Doris parses it, checks each row against the table and the request's settings, and commits all accepted rows in one load transaction. See the Stream Load manual.

The request

Lab 5's first load writes the request out in Python, so that you see every part; the later loads use lab.stream_load(), which sends the same request.

PUT <BE HTTP address>/api/<database>/orders_imported/_stream_load
label: <a name for this batch, kept for retries>
format: csv
column_separator: ,
columns: <the column order of 5.2>
strict_mode: true
max_filter_ratio: 0
group_commit: off_mode
Request body: the bytes of orders.csv

strict_mode = true rejects a row whose value does not convert to its column's type, instead of loading NULL; max_filter_ratio = 0 fails the whole batch if any row is rejected. group_commit = off_mode commits this request on its own (5.4).

Usually a client sends Stream Load to the FE, which answers with a redirect to a BE, and the client follows it, as curl --location-trusted does. The FE redirects to the address that the BE registered. In the course's single-container sandbox, the BE registered 127.0.0.1:8040, its address inside the container, which from your machine is not the BE: the sandbox publishes the BE's port 8040 as 51040. So the Lab sends Stream Load straight to the BE on 51040, and refuses to follow a redirect.

The response, then the table

A load answers with JSON. An HTTP status of 200 only says that the request was received; the JSON says what happened to the rows:

{
  "Status": "Success",
  "NumberLoadedRows": 10,
  "NumberFilteredRows": 0
}

Then query the table, because a load can succeed and still be wrong, with swapped columns for example:

SELECT COUNT(*) AS orders, SUM(order_amount) AS amount
FROM orders_imported;

Expect ten orders and 1400.00. For the WWI history, check also that keys are unique and that order lines, orders, customers, and products join without losing or doubling rows.

A label makes a retry safe, for a while

A label names a load batch in its database. If a request fails or its answer is lost, resend it with the same label: when the first attempt committed, Doris answers Label Already Exists and loads nothing; when it did not, the retry loads the batch once. Lab 5 resends its first file with the same label and checks that the table still holds ten orders.

A label is kept for a limited time. Doris remembers the label of a Stream Load or an INSERT for 12 hours after it finishes (the FE setting streaming_label_keep_max_second), and drops the oldest labels earlier when a database holds more than 2,000 finished loads (label_num_threshold). A retry after that, or with a new label, loads the batch again, and a Duplicate Key table keeps both copies. Retrying a load is therefore not deduplication: removing duplicates of business events needs stable business keys and versions, which Module 7 builds.

A bad row fails the whole batch

Lab 5 then sends two rows with a new label, one with the amount not-a-number. With strict_mode and max_filter_ratio = 0, the response is Fail, NumberFilteredRows is 1, and the table keeps its ten orders and 1400.00: the good row of that batch is not loaded either. ErrorURL points to the rejected row and the reason, for a while; it is a clue for troubleshooting, not a record that you can query later. Module 6 keeps rejected rows in a table.

5.4 Many small loads: Group Commit

Each load transaction has a fixed cost: a commit, and a new version in every tablet it writes, which compaction later has to merge (Module 2). An application that sends one order at a time pays that cost for every order. Group Commit lets Doris collect small loads that arrive close together and commit them as one transaction.

A group commits when it has waited group_commit_interval_ms, a table property of 10,000 ms by default, or when it holds group_commit_data_bytes, 128 MB by default. Lab 5 sets the interval of its table to 2 seconds and sends the ten orders as ten single-row loads, once with off_mode and once with async_mode. With off_mode, ten loads make ten transactions and the tablet's version reaches 11. With async_mode, all ten responses carry the same transaction ID, the tablet reaches version 2, and right after the last response a query still finds none of the ten rows: they appear when the group commits, about 2 seconds later.

Two consequences. First, sync_mode merges only loads that are in flight at the same time: one client that waits for each answer before sending the next gets a commit per load, each taking up to the interval. Second, a Group Commit request cannot carry your label; Doris answers label and group_commit can't be set at the same time. Retrying such a load by label is therefore not possible, and the response's GroupCommit field and its generated label tell you which path the load took. See Group Commit.

5.5 Files in object storage

When the files are already in S3 or another S3-compatible store, Doris can read them there, in two ways:

Fixed files in object storage ── S3 table function ── SELECT: look at the rows
                                                    └─ INSERT INTO ... SELECT: keep them
Fixed files in object storage ── Broker Load job ─── wait for FINISHED, then check the table

The S3 table function reads the files you name as rows of a query; SELECT alone stores nothing, and INSERT INTO ... SELECT writes the rows into a table in the same statement. Lab 4 does exactly this with the lake table's data file. Broker Load submits an asynchronous job: the statement returns at once, and the load is done only when SHOW LOAD reports FINISHED.

External environment example; not executed in the lab. Placeholders in angle brackets stand for your own store; the files and Group Commit extension runs the same job against the course's Silo.

LOAD LABEL <database>.orders_files_202601 (
    DATA INFILE("s3://<bucket>/orders/2026-01/*.parquet")
    INTO TABLE <target_table>
    FORMAT AS "parquet"
)
WITH S3 (
    "AWS_ENDPOINT" = "<endpoint>", "AWS_REGION" = "<region>",
    "AWS_ACCESS_KEY" = "<access_key>", "AWS_SECRET_KEY" = "<secret_key>"
)
PROPERTIES ("timeout" = "3600", "max_filter_ratio" = "0");
SHOW LOAD WHERE Label = 'orders_files_202601';

Before loading the same files again, decide what a second load should do: append, update by key, or replace a range. Appending a month's files to a Duplicate Key table twice counts that month twice. See Broker Load and the S3 table function.

5.6 Two kinds of data, two grains

WWI history in Parquet ── Stream Load ──> wwi_* (ten business tables, 701,846 rows)
Simulated new orders (CSV) ── Stream Load ──> orders_imported (ten orders, practice target)
Raw new orders (JSON, with three errors) ── Stream Load in Module 6 ──> orders_raw ──> orders_clean ──> Module 7

orders_imported is only the practice table for loading, retrying, and rejecting in Lab 5. Module 6 loads the same new orders again, this time with three wrong inputs, into its own raw table and produces orders_clean. Module 7 builds orders_current and order_events from the clean orders; Level 2 starts from that current state and history. The WWI tables are the history: Module 6 reads its customer dimension from wwi_customers and Module 7 its products from wwi_products.

The history package keeps orders, order lines, customers, products, invoices, account transactions, and lookup tables: 701,846 rows in ten tables, which is not a number of orders. Each WWI order line is one product of an order, so COUNT(*) after joining lines to orders counts lines, and a daily report counts orders with COUNT(DISTINCT o.OrderID):

SELECT o.OrderDate,
       COUNT(DISTINCT o.OrderID) AS orders,
       SUM(l.Quantity * l.UnitPrice) AS order_amount
FROM wwi_orders o
JOIN wwi_order_lines l ON o.OrderID = l.OrderID
GROUP BY o.OrderDate
ORDER BY o.OrderDate
LIMIT 10;

WWI's customer transactions are payments and invoices on a customer's account, not on an order: they cannot be split into per-order payments. The further reading on invoices and account receipts compares the three kinds of amounts; it is not part of the Lab. The simulated orders 900001–900010 use WWI's customers and products but their own prices, regions, and times, marked COURSE_SIMULATION.

Lab 5: Batch Loading, Failures, and Retries

Open Lab 5: Batch Loading, Failures, and Retries. You:

1. Load the ten simulated orders with a Stream Load request written out in Python, and check the response and the table: ten orders, 1400.00.
2. Resend the same batch with the same label, and load a batch with a bad amount; the table keeps its ten orders.
3. Send the same ten orders as ten single-row loads, without and with Group Commit, and compare transactions, versions, and visibility.
4. Load the ten WWI history tables from Parquet, and check their keys, relationships, and amounts.
5. Practice: load a CSV whose first two columns are in another order.

You need the course sandbox from Module 1. The Lab does not use the lake environment of Module 4.

Summary

• Choose the load method from where the data is and how it arrives: Stream Load for a finite local file, the S3 table function or Broker Load for files in object storage, Group Commit for many small writes, and Routine Load or CDC for data that keeps arriving.
• A CSV is mapped by position with columns, JSON by key, and Parquet by field name; a default fills only a column that a write leaves out, and never stands in for a business fact.
• Read a load's JSON, not its HTTP status, and then check the table's rows, keys, relationships, and amounts; with strict_mode and max_filter_ratio = 0, one bad row fails the whole batch.
• A label makes a retry of the same batch load nothing twice, for about 12 hours; a new label loads again, and business duplicates need business keys and versions.
• Group Commit merges small loads into fewer transactions and versions; in async_mode the answer comes before the rows are visible, and a Group Commit load cannot carry a label.
• WWI order lines, account transactions, and the simulated orders have different grains and sources: count orders, not lines, and do not force account receipts onto orders.

Quiz 5

Open Quiz 5: Batch and Continuous Data Ingestion. It has six single-choice questions and needs no running environment. The continuous ingestion reading has its own quiz.

Next module: Module 6: Data Quality and Schema Validation.

References

• Apache Doris: Loading overview
• Apache Doris: Stream Load
• Apache Doris: Group Commit
• Apache Doris: S3 table function
• Apache Doris: INSERT INTO SELECT
• Apache Doris: Broker Load
• Apache Doris: Data types