mkpipe-extractor-postgres

PostgreSQL extractor plugin for MkPipe. Reads PostgreSQL tables via JDBC.

Documentation

For more detailed documentation, please visit the GitHub repository.

License

This project is licensed under the Apache 2.0 License - see the LICENSE file for details.

Connection Configuration

connections:
  pg_source:
    variant: postgres
    host: localhost
    port: 5432
    database: mydb
    schema: public
    user: myuser
    password: mypassword

Table Configuration

pipelines:
  - name: pg_to_pg
    source: pg_source
    destination: pg_target
    tables:
      - name: public.users
        target_name: stg_users
        replication_method: full
        fetchsize: 100000

Incremental Replication

      - name: public.users
        target_name: stg_users
        replication_method: incremental
        iterate_column: updated_at
        iterate_column_type: datetime
        partitions_column: id
        partitions_count: 4
        fetchsize: 50000

Custom SQL

      - name: public.orders
        target_name: stg_orders
        replication_method: incremental
        iterate_column: created_at
        iterate_column_type: datetime
        custom_query: "SELECT id, user_id, amount FROM public.orders WHERE {query_filter}"

Use {query_filter} as a placeholder — replaced with the incremental WHERE clause, or WHERE 1=1 on full runs. Can also reference a SQL file:

        custom_query_file: orders.sql   # looks in sql/ directory

Read Parallelism

For large tables, set partitions_column and partitions_count to read in parallel using multiple JDBC connections:

      - name: public.events
        target_name: stg_events
        replication_method: incremental
        iterate_column: created_at
        iterate_column_type: datetime
        partitions_column: id       # numeric column to split on
        partitions_count: 8         # number of parallel JDBC partitions
        fetchsize: 50000

How it works

Spark queries min/max of partitions_column within the time window and divides into partitions_count equal slices
Each slice is fetched by a separate Spark task via a separate JDBC connection
fetchsize controls rows per JDBC round-trip

Performance Notes

Full replication: partitioning is not applied (only incremental).
partitions_column should be a numeric column with good distribution (e.g. primary key).
PostgreSQL default fetchsize in JDBC is 0 (loads all rows at once). Setting fetchsize: 50000–200000 avoids out-of-memory on large tables.

All Table Parameters

Parameter	Type	Default	Description
`name`	string	required	PostgreSQL table name (include schema: `public.users`)
`target_name`	string	required	Destination table name
`replication_method`	`full` / `incremental`	`full`	Replication strategy
`iterate_column`	string or list	—	Column(s) for incremental watermark. String for single column or SQL expression, list for multi-column OR logic
`iterate_column_type`	`int` / `datetime`	—	Type of `iterate_column`
`partitions_column`	string	same as `iterate_column`	Column to split JDBC reads on
`partitions_count`	int	`10`	Number of parallel JDBC partitions
`fetchsize`	int	`100000`	Rows per JDBC fetch
`custom_query`	string	—	Override SQL with `{query_filter}` placeholder
`custom_query_file`	string	—	Path to SQL file (relative to `sql/` dir)
`write_partitions`	int	—	Coalesce to N partitions before writing
`tags`	list	`[]`	Tags for selective pipeline execution
`pass_on_error`	bool	`false`	Skip table on error instead of failing

Metadata

Release files for mkpipe-extractor-postgres 0.5.1

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for mkpipe-extractor-postgres 0.5.1
File	Size	Uploaded	Provenance
mkpipe_extractor_postgres-0.5.1.tar.gz	8.1 kB	Jul 10, 2026	Without provenance	Details

Built distribution (wheel)

Table of built distributions (wheels) for mkpipe-extractor-postgres 0.5.1
File	Interpreter	ABI	Platform	Reset
mkpipe_extractor_postgres-0.5.1-py3-none-any.whl 9.0 kB Jul 10, 2026 Without provenance	Python 3	none	any	Details

Total release size: 17.0 kB

Release files / mkpipe_extractor_postgres-0.5.1.tar.gz

Download URL	mkpipe_extractor_postgres-0.5.1.tar.gz
Size	8.1 kB
Tags	Source
SHA-256 checksum How to use checksums	`a0a9955c9ea7c6d35f55997fd7c6208bc4da4200e971e31d2eba4788b3d2cf9a`
BLAKE2b-256 checksum How to use checksums	`827d589782022c61e15bb6e8d96a04e8ab353c7d0dc0b2154b0dc8efb4886ee2`
Upload date	Jul 10, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`twine/6.2.0 CPython/3.11.15`

Release files / mkpipe_extractor_postgres-0.5.1-py3-none-any.whl

Download URL	mkpipe_extractor_postgres-0.5.1-py3-none-any.whl
Size	9.0 kB
Tags	Python 3
SHA-256 checksum How to use checksums	`d4527677271a4136039605bcff3abf57ca4dba801c15738b7c5de1c1ff08d69e`
BLAKE2b-256 checksum How to use checksums	`6e3c008d1ee44f5231a2433c81cd6db639b6d5bc9fe0c7534ef96a75b23ef0f5`
Upload date	Jul 10, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`twine/6.2.0 CPython/3.11.15`

Security for mkpipe-extractor-postgres 0.5.1

Report project as malware

Provenance

Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.

No provenance found

None of the files in this release were uploaded with provenance metadata.

A note on package security

PyPI is a repository for distributing packages, not a service for auditing them. While third-party security researchers and our staff do their best to detect and remove malicious packages, PyPI cannot guarantee the quality, security or suitability of any given package.

To learn more about protecting yourself from malicious actors, visit our package security documentation.