rivet-cli 0.16.3

Rivet: PostgreSQL/MySQL/SQL Server → Parquet/CSV (local, S3, GCS, Azure). Crate name rivet-cli; binary rivet.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

"""dlt pipeline: extract one Postgres table to local Parquet (Snappy)."""

import os
import sys

import dlt
from dlt.sources.sql_database import sql_table


def run(table_name: str, output_root: str, conn_url: str) -> int:
    os.environ["NORMALIZE__DATA_WRITER__DISABLE_COMPRESSION"] = "true"
    os.environ["DATA_WRITER__FILE_FORMAT"] = "parquet"
    os.environ["NORMALIZE__DATA_WRITER__FILE_FORMAT"] = "parquet"

    table_root = os.path.join(output_root, table_name)
    os.makedirs(table_root, exist_ok=True)

    pipeline = dlt.pipeline(
        pipeline_name=f"bench_{table_name}",
        destination=dlt.destinations.filesystem(bucket_url=f"file://{table_root}"),
        dataset_name="rivet_compete",
        progress=None,
    )
    source = sql_table(
        credentials=conn_url,
        schema="public",
        table=table_name,
        backend="pyarrow",
        chunk_size=10000,
    )
    pipeline.run(source, loader_file_format="parquet")
    return 0


if __name__ == "__main__":
    sys.exit(run(sys.argv[1], sys.argv[2], sys.argv[3]))