diff --git a/full-path-realtime/hits/_s3_ingest/INGEST.md b/full-path-realtime/hits/_s3_ingest/INGEST.md new file mode 100644 index 0000000..ef74181 --- /dev/null +++ b/full-path-realtime/hits/_s3_ingest/INGEST.md @@ -0,0 +1,132 @@ +# Rolling ingest feed for the `hits_100B` dataset + +This sets up a controllable, continuous feed of the `hits_100B` Parquet files +into an S3 bucket you own, at a target ingest rate (rows/sec). It's the shared +source-side piece for full-path real-time ingest tests — any system with an +S3-triggered continuous-ingestion mechanism can point at the resulting bucket: + +- **ClickHouse Cloud** — an S3 ClickPipe in continuous ingestion mode (see + [`clickhouse-cloud/`](clickhouse-cloud/)) +- **Snowflake** — Snowpipe (auto-ingest) watching the same bucket (see + [`snowflake/`](snowflake/)) + +The feeder script (`rolling_s3_copy.py`) does a server-side S3-to-S3 copy — +data never transits through the machine running the script, so it's fast and +doesn't burn local disk or bandwidth. + +--- + +## 1. Prerequisites + +- An AWS account/role with: + - `s3:GetObject` / `s3:ListBucket` on the source bucket (`public-pme` — already + public-read, so this works with no extra setup) + - `s3:PutObject` on your destination bucket +- AWS credentials configured on the machine running the script (`aws configure`, + or an instance role if running on EC2) +- Python 3.8+ + +## 2. Set up the ingester environment + +```bash +# 1. Make sure venv support is installed +sudo apt-get update +sudo apt-get install -y python3-venv + +# 2. Create a venv for the ingester +python3 -m venv ~/.venvs/ch-ingest + +# 3. Activate it +source ~/.venvs/ch-ingest/bin/activate + +# 4. Install dependencies +pip install --upgrade pip +pip install boto3 clickhouse-connect pyarrow +pip install thriftpy2 +``` + +`boto3` is required by `rolling_s3_copy.py` itself (server-side S3 copy). +`clickhouse-connect`, `pyarrow`, and `thriftpy2` are for validation/analysis +scripts elsewhere in this benchmark (e.g. querying ClickHouse Cloud or reading +Parquet metadata directly) — install them here too so the same venv covers +both feeding and checking the ingest. + +Remember to `source ~/.venvs/ch-ingest/bin/activate` again in any new shell +before running these scripts. + +## 3. Create the destination bucket + +Same region as the source (`eu-west-3`) avoids cross-region transfer fees and +is faster: + +```bash +aws s3 mb s3://my-benchmark-bucket --region eu-west-3 +``` + +Grant read access on this bucket to whichever system will be watching it +(ClickPipes IAM role, Snowflake's storage integration role, etc.) — see the +relevant system folder for the exact policy. + +## 4. Run the feeder + +`rolling_s3_copy.py` lives alongside this doc. It paces file copies so the +destination bucket fills at a target rows/sec rate: + +``` +interval_between_files = rows_per_file (~50,000,000) / eps +``` + +Sanity-check the timing first, without copying anything: + +```bash +./rolling_s3_copy.py --dest-bucket my-benchmark-bucket --eps 1000000 --dry-run +``` + +Run for real, at the dataset's native rate (1,000,000 rows/sec): + +```bash +./rolling_s3_copy.py --dest-bucket my-benchmark-bucket --eps 1000000 +``` + +Run slower, e.g. 100,000 rows/sec: + +```bash +./rolling_s3_copy.py --dest-bucket my-benchmark-bucket --eps 100000 +``` + +Resume after an interruption (pick up at file index 500): + +```bash +./rolling_s3_copy.py --dest-bucket my-benchmark-bucket --eps 1000000 --start-index 500 +``` + +Loop indefinitely once all 2,000 files are copied (keeps keys lexicographically +increasing across passes, which matters if the consumer uses ordered +continuous-ingestion mode): + +```bash +./rolling_s3_copy.py --dest-bucket my-benchmark-bucket --eps 1000000 --loop +``` + +Run `./rolling_s3_copy.py --help` for the full flag list. + +## 5. Point a consumer at the bucket + +Once the feeder is running, configure the system under test to continuously +ingest from `my-benchmark-bucket`: + +- ClickHouse Cloud: see [`clickhouse-cloud/`](clickhouse-cloud/) for the + S3 ClickPipe setup (continuous ingestion mode). +- Snowflake: see [`snowflake/`](snowflake/) for the Snowpipe auto-ingest setup + (S3 event notifications → SQS → Snowpipe). + +## 6. Notes + +- The feeder does a managed multipart server-side copy per file (~8.78 GB + median), so per-file copy time is typically well under the interval needed + for realistic eps targets. If you see "falling behind eps target" warnings, + either raise `--eps` expectations or check for throttling on the destination + bucket. +- No local disk space is required by the feeder itself — only the destination + S3 bucket needs to be able to hold whatever portion of the ~17.5 TB dataset + you plan to feed through. diff --git a/full-path-realtime/hits/_s3_ingest/_call.txt b/full-path-realtime/hits/_s3_ingest/_call.txt new file mode 100644 index 0000000..ed8c193 --- /dev/null +++ b/full-path-realtime/hits/_s3_ingest/_call.txt @@ -0,0 +1,43 @@ +https://s3.eu-west-3.amazonaws.com/public-pme/hits_100B/2000x50m_span_27h46m40s_rps_1m/hits_p*.parquet + +https://s3.eu-west-3.amazonaws.com/public-pme/hits_100B_feed/hits_p*.parquet + +https://s3.eu-west-3.amazonaws.com/public-pme/hits_100B/2000x50m_span_27h46m40s_rps_1m/hits_p00000.parquet + + + +- (1) clear folder +aws s3 rm s3://public-pme/hits_100B_feed/ --recursive + +- (2) seed folder +aws s3 cp \ + s3://public-pme/hits_100B/2000x50m_span_27h46m40s_rps_1m/hits_p00000.parquet \ + s3://public-pme/hits_100B_feed/hits_p00000.parquet + +- (3) Create the pipeline - initial load ingests the real 50M rows from file 0
- wait until that happened... + +- (4) Run for 27 hours +./rolling_s3_copy.py \ + --dest-bucket public-pme \ + --dest-prefix hits_100B_feed/ \ + --eps 1000000 \ + --start-index 1 + + + + + + + + + + + + + +./rolling_s3_copy.py \ + --dest-bucket public-pme \ + --dest-prefix hits_100B_feed/ \ + --eps 1000000 \ + --dry-run + diff --git a/full-path-realtime/hits/_s3_ingest/rolling_s3_copy.py b/full-path-realtime/hits/_s3_ingest/rolling_s3_copy.py new file mode 100755 index 0000000..8608f8c --- /dev/null +++ b/full-path-realtime/hits/_s3_ingest/rolling_s3_copy.py @@ -0,0 +1,226 @@ +#!/usr/bin/env python3 +""" +rolling_s3_copy.py + +Continuously copies the ClickHouse "hits_100B" benchmark Parquet files from the +public source bucket into a destination bucket/prefix you control, pacing the +copies so the *effective ingest rate* (rows/sec, i.e. "eps") matches a target +you set. + +This does a server-side S3-to-S3 copy (via boto3's managed multipart copy) -- +data never transits through this machine, so it's fast and doesn't burn your +instance's bandwidth or local disk. + +Why pacing by file works: + Each source file holds ~50,000,000 rows. If you want the destination + prefix to fill up at a rate of `--eps` rows/sec, you copy one file every + + interval = rows_per_file / eps seconds + + e.g. eps=1,000,000 (the dataset's native rate) -> ~50s between files, + which reproduces the original real-time cadence. eps=100,000 -> ~500s + (8m20s) between files, i.e. 10x slower than native. + +Requirements: + pip install boto3 --break-system-packages # or use a venv + AWS credentials configured (aws configure) with: + - GetObject/ListBucket on the source bucket + - PutObject on your destination bucket/prefix + +Examples: + # Dry run to sanity check timing without copying anything: + ./rolling_s3_copy.py --dest-bucket public-pme --dest-prefix hits_100B_feed/ \\ + --eps 1000000 --dry-run + + # Real run at native rate (1,000,000 eps), logging to a file too: + ./rolling_s3_copy.py --dest-bucket public-pme --dest-prefix hits_100B_feed/ \\ + --eps 1000000 --log-file ./rolling_copy.log + + # 10x slower than native, resuming from file 500 after an interruption: + ./rolling_s3_copy.py --dest-bucket public-pme --dest-prefix hits_100B_feed/ \\ + --eps 100000 --start-index 500 + + # Loop indefinitely, keeping every file's key lexicographically increasing + # across passes (needed for ClickPipes' default lexicographic + # continuous-ingestion mode): + ./rolling_s3_copy.py --dest-bucket public-pme --dest-prefix hits_100B_feed/ \\ + --eps 1000000 --loop +""" + +import argparse +import logging +import sys +import time + +import boto3 +from boto3.s3.transfer import TransferConfig + +DEFAULT_SOURCE_BUCKET = "public-pme" +DEFAULT_SOURCE_PREFIX = "hits_100B/2000x50m_span_27h46m40s_rps_1m/" +DEFAULT_ROWS_PER_FILE = 50_000_000 # README median is 49,999,863; close enough +DEFAULT_NUM_FILES = 2000 +SUMMARY_EVERY = 20 # print a running-average summary every N files + +log = logging.getLogger("rolling_s3_copy") + + +def setup_logging(log_file=None): + log.setLevel(logging.INFO) + fmt = logging.Formatter("%(asctime)s [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S") + + stream_handler = logging.StreamHandler(sys.stdout) + stream_handler.setFormatter(fmt) + log.addHandler(stream_handler) + + if log_file: + file_handler = logging.FileHandler(log_file) + file_handler.setFormatter(fmt) + log.addHandler(file_handler) + + +def human_bytes(n): + for unit in ("B", "KiB", "MiB", "GiB", "TiB"): + if n < 1024: + return f"{n:.2f} {unit}" + n /= 1024 + return f"{n:.2f} PiB" + + +def parse_args(): + p = argparse.ArgumentParser( + description="Pace a rolling S3-to-S3 copy of the hits_100B dataset to simulate a target ingest rate." + ) + p.add_argument("--source-bucket", default=DEFAULT_SOURCE_BUCKET) + p.add_argument("--source-prefix", default=DEFAULT_SOURCE_PREFIX) + p.add_argument("--dest-bucket", required=True) + p.add_argument("--dest-prefix", default="", help="Optional key prefix in the destination bucket") + p.add_argument("--num-files", type=int, default=DEFAULT_NUM_FILES) + p.add_argument("--rows-per-file", type=int, default=DEFAULT_ROWS_PER_FILE) + p.add_argument("--eps", type=float, required=True, help="Target ingest rate in rows/sec") + p.add_argument("--start-index", type=int, default=0, help="File index to start/resume from (0-1999)") + p.add_argument("--region", default="eu-west-3", help="Region for the S3 client") + p.add_argument("--loop", action="store_true", help="After the last file, wrap around and keep going") + p.add_argument("--dry-run", action="store_true", help="Print the plan without copying anything") + p.add_argument("--log-file", default=None, help="Also write logs to this file (in addition to stdout)") + return p.parse_args() + + +def main(): + args = parse_args() + setup_logging(args.log_file) + + if args.eps <= 0: + sys.exit("--eps must be positive") + + interval = args.rows_per_file / args.eps + total_files = args.num_files - args.start_index + log.info( + "Target eps: %s rows/sec | rows/file: %s | interval: %.2fs | files: %d (index %d-%d)%s | " + "single-pass wall time: ~%.2fh", + f"{args.eps:,.0f}", + f"{args.rows_per_file:,}", + interval, + total_files, + args.start_index, + args.num_files - 1, + ", looping indefinitely after that" if args.loop else "", + total_files * interval / 3600, + ) + + if args.dry_run: + log.info("Dry run only -- no objects will be copied.") + return + + s3 = boto3.client("s3", region_name=args.region) + transfer_config = TransferConfig(multipart_threshold=1024 * 1024 * 1024, max_concurrency=10) + + # Running totals for periodic throughput summaries. + run_start = time.time() + files_done = 0 + bytes_done = 0 + rows_done = 0 + + generation = 0 + while True: + for idx in range(args.start_index, args.num_files): + src_key = f"{args.source_prefix}hits_p{idx:05d}.parquet" + + if args.loop and generation > 0: + # Keep keys lexicographically increasing across passes so + # ClickPipes' default ordered-ingestion mode keeps working. + dest_key = f"{args.dest_prefix}g{generation:04d}_hits_p{idx:05d}.parquet" + else: + dest_key = f"{args.dest_prefix}hits_p{idx:05d}.parquet" + + try: + size_bytes = s3.head_object(Bucket=args.source_bucket, Key=src_key)["ContentLength"] + except Exception as exc: + log.warning("Could not stat %s before copying (%s) -- continuing anyway", src_key, exc) + size_bytes = None + + log.info( + "[gen %d %d/%d] copying %s -> s3://%s/%s%s", + generation, + idx + 1, + args.num_files, + src_key, + args.dest_bucket, + dest_key, + f" ({human_bytes(size_bytes)})" if size_bytes else "", + ) + + t0 = time.time() + s3.copy( + {"Bucket": args.source_bucket, "Key": src_key}, + args.dest_bucket, + dest_key, + Config=transfer_config, + ) + elapsed = time.time() - t0 + + throughput = f"{human_bytes(size_bytes / elapsed)}/s" if size_bytes and elapsed > 0 else "n/a" + log.info( + "[gen %d %d/%d] done in %.1fs (throughput: %s)", + generation, + idx + 1, + args.num_files, + elapsed, + throughput, + ) + + files_done += 1 + if size_bytes: + bytes_done += size_bytes + rows_done += args.rows_per_file + + if files_done % SUMMARY_EVERY == 0: + wall_elapsed = time.time() - run_start + avg_throughput = bytes_done / wall_elapsed if wall_elapsed > 0 else 0 + avg_eps = rows_done / wall_elapsed if wall_elapsed > 0 else 0 + log.info( + "-- summary: %d files copied, %s total, running avg %s/s, %.0f rows/sec (target %.0f) --", + files_done, + human_bytes(bytes_done), + human_bytes(avg_throughput), + avg_eps, + args.eps, + ) + + sleep_time = interval - elapsed + if sleep_time > 0: + time.sleep(sleep_time) + else: + log.warning( + "copy took %.1fs, longer than the %.1fs target interval -- falling behind eps target", + elapsed, + interval, + ) + + if not args.loop: + break + generation += 1 + args.start_index = 0 # subsequent passes always start from file 0 + + +if __name__ == "__main__": + main() diff --git a/full-path-realtime/hits/clickhouse-cloud/create.sql b/full-path-realtime/hits/clickhouse-cloud/create.sql new file mode 100644 index 0000000..5297c61 --- /dev/null +++ b/full-path-realtime/hits/clickhouse-cloud/create.sql @@ -0,0 +1,109 @@ +CREATE TABLE hits +( + WatchID BIGINT NOT NULL, + JavaEnable SMALLINT NOT NULL, + Title TEXT NOT NULL, + GoodEvent SMALLINT NOT NULL, + EventTime DateTime64(6, 'UTC') NOT NULL, + EventDate Date NOT NULL, + CounterID INTEGER NOT NULL, + ClientIP INTEGER NOT NULL, + RegionID INTEGER NOT NULL, + UserID BIGINT NOT NULL, + CounterClass SMALLINT NOT NULL, + OS SMALLINT NOT NULL, + UserAgent SMALLINT NOT NULL, + URL TEXT NOT NULL, + Referer TEXT NOT NULL, + IsRefresh SMALLINT NOT NULL, + RefererCategoryID SMALLINT NOT NULL, + RefererRegionID INTEGER NOT NULL, + URLCategoryID SMALLINT NOT NULL, + URLRegionID INTEGER NOT NULL, + ResolutionWidth SMALLINT NOT NULL, + ResolutionHeight SMALLINT NOT NULL, + ResolutionDepth SMALLINT NOT NULL, + FlashMajor SMALLINT NOT NULL, + FlashMinor SMALLINT NOT NULL, + FlashMinor2 TEXT NOT NULL, + NetMajor SMALLINT NOT NULL, + NetMinor SMALLINT NOT NULL, + UserAgentMajor SMALLINT NOT NULL, + UserAgentMinor VARCHAR(255) NOT NULL, + CookieEnable SMALLINT NOT NULL, + JavascriptEnable SMALLINT NOT NULL, + IsMobile SMALLINT NOT NULL, + MobilePhone SMALLINT NOT NULL, + MobilePhoneModel TEXT NOT NULL, + Params TEXT NOT NULL, + IPNetworkID INTEGER NOT NULL, + TraficSourceID SMALLINT NOT NULL, + SearchEngineID SMALLINT NOT NULL, + SearchPhrase TEXT NOT NULL, + AdvEngineID SMALLINT NOT NULL, + IsArtifical SMALLINT NOT NULL, + WindowClientWidth SMALLINT NOT NULL, + WindowClientHeight SMALLINT NOT NULL, + ClientTimeZone SMALLINT NOT NULL, + ClientEventTime DateTime64(3, 'UTC') NOT NULL, + SilverlightVersion1 SMALLINT NOT NULL, + SilverlightVersion2 SMALLINT NOT NULL, + SilverlightVersion3 INTEGER NOT NULL, + SilverlightVersion4 SMALLINT NOT NULL, + PageCharset TEXT NOT NULL, + CodeVersion INTEGER NOT NULL, + IsLink SMALLINT NOT NULL, + IsDownload SMALLINT NOT NULL, + IsNotBounce SMALLINT NOT NULL, + FUniqID BIGINT NOT NULL, + OriginalURL TEXT NOT NULL, + HID INTEGER NOT NULL, + IsOldCounter SMALLINT NOT NULL, + IsEvent SMALLINT NOT NULL, + IsParameter SMALLINT NOT NULL, + DontCountHits SMALLINT NOT NULL, + WithHash SMALLINT NOT NULL, + HitColor CHAR NOT NULL, + LocalEventTime DateTime64(3, 'UTC') NOT NULL, + Age SMALLINT NOT NULL, + Sex SMALLINT NOT NULL, + Income SMALLINT NOT NULL, + Interests SMALLINT NOT NULL, + Robotness SMALLINT NOT NULL, + RemoteIP INTEGER NOT NULL, + WindowName INTEGER NOT NULL, + OpenerName INTEGER NOT NULL, + HistoryLength SMALLINT NOT NULL, + BrowserLanguage TEXT NOT NULL, + BrowserCountry TEXT NOT NULL, + SocialNetwork TEXT NOT NULL, + SocialAction TEXT NOT NULL, + HTTPError SMALLINT NOT NULL, + SendTiming INTEGER NOT NULL, + DNSTiming INTEGER NOT NULL, + ConnectTiming INTEGER NOT NULL, + ResponseStartTiming INTEGER NOT NULL, + ResponseEndTiming INTEGER NOT NULL, + FetchTiming INTEGER NOT NULL, + SocialSourceNetworkID SMALLINT NOT NULL, + SocialSourcePage TEXT NOT NULL, + ParamPrice BIGINT NOT NULL, + ParamOrderID TEXT NOT NULL, + ParamCurrency TEXT NOT NULL, + ParamCurrencyID SMALLINT NOT NULL, + OpenstatServiceName TEXT NOT NULL, + OpenstatCampaignID TEXT NOT NULL, + OpenstatAdID TEXT NOT NULL, + OpenstatSourceID TEXT NOT NULL, + UTMSource TEXT NOT NULL, + UTMMedium TEXT NOT NULL, + UTMCampaign TEXT NOT NULL, + UTMContent TEXT NOT NULL, + UTMTerm TEXT NOT NULL, + FromTag TEXT NOT NULL, + HasGCLID SMALLINT NOT NULL, + RefererHash BIGINT NOT NULL, + URLHash BIGINT NOT NULL, + CLID INTEGER NOT NULL, + PRIMARY KEY (CounterID, EventDate, UserID, EventTime, WatchID) +); \ No newline at end of file