joinem 0.14.0

joinem provides a CLI for fast, flexbile concatenation of tabular data using polars

• Free software: MIT license
• Repository: https://github.com/mmore500/joinem
• Documentation: https://github.com/mmore500/joinem/blob/master/README.md

Install

python3 -m pip install joinem

Features

• Lazily streams I/O to expeditiously handle numerous large files.
• Supports CSV, JSON, parquet, and feather input files.
  • JSON input must be newline-delimited (.json, .jsonl, or .ndjson).
  • Compressed input is supported: .csv.gz, .csv.zst, .json.gz, .json.zst, .jsonl.gz, .jsonl.zst, .ndjson.gz, and .ndjson.zst.
  • Feather files are read eagerly rather than streamed.
  • Input formats may be mixed.
• Supports CSV, JSON, parquet, and feather output files.
  • JSON output is newline-delimited (.json, .jsonl, or .ndjson).
  • Compressed output is supported: .csv.gz, .csv.zst, .json.gz, .json.zst, .jsonl.gz, .jsonl.zst, .ndjson.gz, and .ndjson.zst (requires a recent polars).
• Allows mismatched columns and/or empty data files with --how diagonal and --how diagonal_relaxed.
• Logs bytes written and write throughput with --progress.
• Add programatically-generated columns to output.

Example Usage

Pass input filenames via stdin, one filename per line.

find path/to/*.parquet path/to/*.csv | python3 -m joinem out.parquet

Output file type is inferred from the extension of the output file name. Supported output types are feather, JSON, parquet, and csv.

find -name '*.parquet' | python3 -m joinem out.json

If file columns may mismatch, use --how diagonal.

find path/to/ -name '*.csv' | python3 -m joinem out.csv --how diagonal

If some files may be empty, use --how diagonal_relaxed.

To run via Singularity/Apptainer,

ls -1 *.csv | singularity run docker://ghcr.io/mmore500/joinem:v0.14.0 out.feather

Advanced Usage

Add literal value column to output.

ls -1 *.csv | python3 -m joinem out.csv --with-column 'pl.lit(2).alias("two")'

Cast a column to categorical in the output, shrink dtypes, and tune compression.

ls -1 *.csv | python3 -m joinem out.pqt \
  --with-column 'pl.col("uuid").cast(pl.Categorical)' --string-cache \
  --shrink-dtypes \
  --write-kwarg 'compression_level=10' --write-kwarg 'compression="zstd"'

Alias an existing column in the output.

ls -1 *.csv | python3 -m joinem out.csv --with-column 'pl.col("a").alias("a2")'

Apply regex on source datafile paths to create new column in output.

ls -1 path/to/*.csv | python3 -m joinem out.csv \
  --with-column 'pl.lit(filepath).str.replace(r".*?([^/]*)\.csv", r"${1}").alias("filename stem")'

Write gzip-compressed CSV output; compression is inferred from the output file extension.

ls -1 *.csv.gz | python3 -m joinem out.csv.gz

Read data from stdin and write data to stdout.

cat foo.csv | python3 -m joinem "/dev/stdout" --stdin \
  --output-filetype csv --input-filetype csv

Write to parquet via stdout using pv to display progress, cast "myValue" column to categorical, and use lz4 for parquet compression.

ls -1 input/*.pqt | python3 -m joinem "/dev/stdout" --output-filetype pqt \
  --with-column 'pl.col("myValue").cast(pl.Categorical)' \
  --write-kwarg 'compression="lz4"' \
  | pv > concat.pqt

Use --init-eval to configure Polars before processing begins, e.g., to set pl.Config.set_engine_affinity to prefer the streaming engine.

ls -1 *.csv | python3 -m joinem out.pqt \
  --init-eval 'pl.Config.set_engine_affinity(engine="streaming")'

API

usage: __main__.py [-h] [--version] [--progress] [--stdin] [--drop DROP]
                   [--select SELECT] [--eager-read] [--eager-write]
                   [--filter FILTERS] [--head HEAD] [--tail TAIL]
                   [--gather-every GATHER_EVERY] [--sample SAMPLE]
                   [--shuffle] [--seed SEED] [--init-eval INIT_EVALS]
                   [--with-column WITH_COLUMNS] [--shrink-dtypes]
                   [--string-cache]
                   [--how {vertical,vertical_relaxed,diagonal,diagonal_relaxed,horizontal,align,align_full,align_inner,align_left,align_right}]
                   [--input-filetype INPUT_FILETYPE]
                   [--output-filetype OUTPUT_FILETYPE]
                   [--read-kwarg READ_KWARGS] [--write-kwarg WRITE_KWARGS]
                   output_file

CLI for fast, flexbile concatenation of tabular data using Polars.

positional arguments:
  output_file           Output file name

options:
  -h, --help            show this help message and exit
  --version             show program's version number and exit
  --progress            Log bytes written and write throughput.
  --stdin               Read data from stdin.
  --drop DROP           Column names to drop. Flag may be repeated to
                        provide multiple column names.
  --select SELECT       Column names to select; otherwise, all columns are
                        selected. Flag may be repeated to provide multiple
                        column names.
  --eager-read          Use read_* instead of scan_*. Can improve
                        performance in some cases.
  --eager-write         Use write_* instead of sink_*. Can improve
                        performance in some cases.
  --filter FILTERS      Expression to be evaluated and passed to polars
                        DataFrame.filter. Example: 'pl.col("thing") == 0'
  --head HEAD           Number of rows to include in output, counting from
                        front.
  --tail TAIL           Number of rows to include in output, counting from
                        back.
  --gather-every GATHER_EVERY
                        Take every nth row.
  --sample SAMPLE       Number of rows to include in output, sampled
                        uniformly. Accepts an integer or polars expression,
                        with access to polars as `pl`.
                        Sampling more rows than available raises an error;
                        to sample up to a maximum, use an expression such as
                        'pl.len().clip(upper_bound=100)'. Pass --seed for
                        deterministic behavior.
  --shuffle             Should output be shuffled? Pass --seed for
                        deterministic behavior.
  --seed SEED           Integer seed for deterministic behavior.
  --init-eval INIT_EVALS
                        Expression to be evaluated once for initialization
                        side effects before any data is processed, has
                        access to polars as `pl`. Flag may be repeated to
                        provide multiple expressions. Example:
                        'pl.Config.set_engine_affinity(engine="streaming")'
  --with-column WITH_COLUMNS
                        Expression to be evaluated to add a column, has
                        access to each datafile's filepath as `filepath` and
                        polars as `pl`. Flag may be repeated to provide
                        multiple expressions. Example:
                        'pl.lit(filepath).str.replace(r".*?([^/]*)\.csv",
                        r"${1}").alias("filename stem")'
  --shrink-dtypes       Shrink numeric columns to the minimal required
                        datatype.
  --string-cache        Enable Polars global string cache.
  --how {vertical,vertical_relaxed,diagonal,diagonal_relaxed,horizontal,align,align_full,align_inner,align_left,align_right}
                        How to concatenate frames. See
                        <https://docs.pola.rs/py-
                        polars/html/reference/api/polars.concat.html> for
                        more information.
  --input-filetype INPUT_FILETYPE
                        Filetype of input. Otherwise, inferred. Example:
                        csv, parquet, json, feather
  --output-filetype OUTPUT_FILETYPE
                        Filetype of output. Otherwise, inferred. Example:
                        csv, parquet
  --read-kwarg READ_KWARGS
                        Additional keyword arguments to pass to pl.read_* or
                        pl.scan_* call(s). Provide as 'key=value'. Specify
                        multiple kwargs by using this flag multiple times.
                        Arguments will be evaluated as Python expressions.
                        Example: 'infer_schema_length=None'
  --write-kwarg WRITE_KWARGS
                        Additional keyword arguments to pass to pl.write_*
                        or pl.sink_* call. Provide as 'key=value'. Specify
                        multiple kwargs by using this flag multiple times.
                        Arguments will be evaluated as Python expressions.
                        Example: 'compression="lz4"'

Provide input filepaths via stdin. Example: find path/to/ -name '*.csv' |
python3 -m joinem out.csv

Citing

If joinem contributes to a scholarly work, please cite it as

Matthew Andres Moreno. (2024). mmore500/joinem. Zenodo. https://doi.org/10.5281/zenodo.10701182

@software{moreno2024joinem,
  author = {Matthew Andres Moreno},
  title = {mmore500/joinem},
  month = feb,
  year = 2024,
  publisher = {Zenodo},
  doi = {10.5281/zenodo.10701182},
  url = {https://doi.org/10.5281/zenodo.10701182}
}

And don't forget to leave a star on GitHub!