FIX CSV import

[marvin-hansen]

I tried to import a dataset comprising of over 700 CSV files totallng a billion rows.
Despite best effort and despite all the help from the engineering team in the slack channel, it's basically impossible.

How to reproduce

1. Download dataset
   https://support.kraken.com/hc/en-us/articles/360047543791-Downloadable-historical-market-data-time-and-sales-

2. Select the top ten largest CSV files:

du -ah | sort -rh | head -n 10
31G .
2.6G ./XBTEUR.csv
2.1G ./XBTUSD.csv
1.6G ./USDTUSD.csv
1.6G ./USDTEUR.csv
1.4G ./ETHUSD.csv
1.2G ./ETHEUR.csv
1.1G ./ETHXBT.csv
663M ./XRPEUR.csv
628M ./EURUSD.csv

Pick any of them, say XBTEUR, create a table:

CREATE STREAM default.kraken_xbteur
(
  `timestamp` datetime64(3),
  `price` float64,
  `volume` float64,
  `_tp_time` datetime64(3, 'UTC') DEFAULT timestamp CODEC(DoubleDelta, LZ4),
  INDEX _tp_time_index _tp_time TYPE minmax GRANULARITY 2
)
ENGINE = Stream(1, 1, rand())
PARTITION BY to_YYYYMM(timestamp)
ORDER BY to_start_of_hour(_tp_time)
SETTINGS event_time_column = 'timestamp', index_granularity = 8192

Try to import into the newly created table:

INSERT INTO kraken_xbteur (timestamp,price,volume) SELECT timestamp,price,volume FROM file('XBTEUR.csv', 'CSV', 'timestamp datetime64(3), price float64, volume float64')  

Error message and/or stacktrace

There are multiple problems, as discussed in Slack.

1. File size limit is set to 100 MB. This is ridiculous. Just fix this in the next release and set it to a more realistic 15 GB.
2. Starting the proton server with a custom configuration fails with multiple errors due to missing env variable and incorrect file permission.

If you use the Rust client to insert batch-wise, you encounter another error related to the size limit regardless of batch-size.

Additional context

When this is solved, please publish benchmarks that demonstrates that Proton can handle larger imports.

Specifically, for any serious application proton must demonstrate that it can import the following datasets:

• 1.1 Billion Taxi Rides Benchmarks
• 11.5 billion rows/ticks Forex

In finance, you want to insert all historical trade data into a table, connect to a live stream, and then keep filling the table to keep the trade data up to date. It's self evident that this cannot be done with Proton until large CSV file import gets fixed. And even then I want to see real import benchmarks because I noticed some performance problems.

[chenziliang]

Thanks @marvin-hansen for reporting this issue. Tackling it. ETA by today PST.

[chenziliang]

And also would like to share some tech details regarding proton.

1. The by default setting in Proton is optimized for real-time / high concurrent / small batch streaming data. So by default, the stream schema is not optimized for large amount of historical data loading. Users need refine the partition by CREATE STREAM clause to better fit historical data loading which spans across long time (decade in your case) to avoid too many partitions error. We will need document / inform users better on this.
2. Proton is by default handling real-time data with event timestamp and optimized for that (_tp_time). We will need handle reserved columns in Proton smoothly for CSV etc bulk historical data loading.
3. Proton optimizes for high frequent small batch ingest (1 or a dozens of events per insert / ingest) instead of GB CSV file bulk loading. Having said this, we will fix Proton to work with GB CSV file bulking loading.

[chenziliang]

May I ask after importing the large amount of historical data, what kind of processing you like to explore and how you connect this historical data to live stream etc?

[jovezhong]

Hi @marvin-hansen , while @chenziliang is working a better handling for large amount of CSV files in Proton, I managed to load all 700 CSV files with the current Proton 1.5.1.

Long story short, the SQL is

INSERT INTO kraken_all (path,timestamp,price,volume) 
SELECT _path,timestamp,price,volume 
FROM file('data/*.csv', 'CSV', 'timestamp datetime64(3), price float64, volume float64')

Elapsed: 844.557 sec. Processed 989.90 million rows, 24.75 GB (1.17 million rows/s., 29.30 MB/s.)

select count() from table(kraken_all)
┌───count()─┐
│ 988931335 │
└───────────┘

ls -1 | wc -l
708

Happy to share more in our community slack. Key steps:

• curl https://install.timeplus.com | sh
• sudo ./proton install
• sudo chmod +w /etc/proton-server/config.yaml
• sudo vim /etc/proton-server/config.yaml to change log_max_record_size under nativelog (10485760 ->90000000)
• proton stop
• proton start
• sudo cp -rf data /var/lib/proton/user_files
• create stream

CREATE STREAM kraken_all
(
 `path` string,
  `timestamp` datetime64(3),
  `price` float64,
  `volume` float64,
  `_tp_time` datetime64(3, 'UTC') DEFAULT timestamp CODEC(DoubleDelta, LZ4),
  INDEX _tp_time_index _tp_time TYPE minmax GRANULARITY 2
)
ENGINE = Stream(1, 1, rand())
PARTITION BY to_YYYYMM(_tp_time)
ORDER BY to_start_of_hour(_tp_time)
SETTINGS event_time_column = 'timestamp', index_granularity = 8192;

Then insert all files with _path as virtual column to capture the file name (you may need the symbol in the file name)

INSERT INTO kraken_all (path,timestamp,price,volume) SELECT _path,timestamp,price,volume FROM file('data/*.csv', 'CSV', 'timestamp datetime64(3), price float64, volume float64')

It took me 800 seconds to load that 1billion rows. We certainly can improve the performance. Currently data goes to Proton NativeLog (stream storage) first. If you just need to query them in the batch way, we may use ClickHouse's engine to skip the stream storage.

A screenshot I took during the ingestion (half way)

[chenziliang]

With the fix, users shall be able to load the CSV files as

CREATE STREAM kraken(
    `timestamp` datetime64(3) CODEC(DoubleDelta, LZ4),
    `price` float64 CODEC(Delta, LZ4),
    `volume` float64 CODEC(Delta, LZ4)
) 
 PARTITION BY to_YYYYMM(timestamp)
SETTINGS logstore_retention_bytes=10737418240;

-- Non-interactive insert
./proton client -q "INSERT INTO default.kraken(* except(_tp_time)) FORMAT CSV" < /code/trade/XBTEUR.csv

-- Interactive insert
timeplus :) INSERT INTO default.kraken(* except(_tp_time)) FROM INFILE '/code/trade/XBTEUR.csv' FORMAT CSV;

INSERT INTO default.kraken (* EXCEPT _tp_time) FROM INFILE /code/trade/XBTEUR.csv FORMAT CSV
Query id: 67a9f968-d5c1-4a05-a44f-b8270e5ef6e1

Ok.

83059750 rows in set. Elapsed: 4.957 sec. 

-- INSERT AS SELECT as above

timeplus :) INSERT INTO kraken(path, timestamp,price,volume) 
SELECT _path,timestamp,price,volume 
FROM file('data/*.csv', 'CSV', 'timestamp datetime64(3), price float64, volume float64') 
SETTINGS max_insert_threads=8;

[chenziliang]

For non-interactive, interactive bulk loading, we support wildcard match as well. For example

./proton client -q "INSERT INTO default.kraken(* except(_tp_time)) FORMAT CSV" < /code/trade/*.csv
timeplus :) INSERT INTO default.kraken(* except(_tp_time)) FROM INFILE '/code/trade/*.csv' FORMAT CSV;

[chenziliang]

Please note Proton is designed for concurrent ingest, the current serial ingest may have some perf challenges compared a pure OLAP system like ClickHouse.

[chenziliang]

With the enhancement, the by default setting shall just work which means wget the Proton binary, launch it as proton server without a customized config and then load the data.

[chenziliang]

If file system is not a bottleneck, we can create multi-shards stream to enable more concurrency, for example

CREATE STREAM kraken(
    `timestamp` datetime64(3) CODEC(DoubleDelta, LZ4),
    `price` float64 CODEC(Delta, LZ4),
    `volume` float64 CODEC(Delta, LZ4)
) 
 PARTITION BY to_YYYYMM(timestamp)
SETTINGS shards=3, logstore_retention_bytes=10737418240;