Skip to main content

Load data from databases to dataframes, the fastest way.

Project description

ConnectorX status docs

Load data from to , the fastest way.

For more data sources, please check out our discussion.

ConnectorX enables you to load data from databases into Python in the fastest and most memory efficient way.

What you need is one line of code:

import connectorx as cx

cx.read_sql("postgresql://username:password@server:port/database", "SELECT * FROM lineitem")

Optionally, you can accelerate the data loading using parallelism by specifying a partition column.

import connectorx as cx

cx.read_sql("postgresql://username:password@server:port/database", "SELECT * FROM lineitem", partition_on="l_orderkey", partition_num=10)

The function will partition the query by evenly splitting the specified column to the amount of partitions. ConnectorX will assign one thread for each partition to load and write data in parallel. Currently, we support partitioning on integer columns for SPJA queries.

Check out more detailed usage and examples here.

Installation

pip install connectorx

Performance

We compared different solutions in Python that provides the read_sql function, by loading a 10x TPC-H lineitem table (8.6GB) from Postgres into a DataFrame, with 4 cores parallelism.

Time chart, lower is better.

time chart

Memory consumption chart, lower is better.

memory chart

In conclusion, ConnectorX uses up to 3x less memory and 21x less time. More on here.

How does ConnectorX achieve a lightening speed while keeping the memory footprint low?

We observe that existing solutions more or less do data copy multiple times when downloading the data. Additionally, implementing a data intensive application in Python brings additional cost.

ConnectorX is written in Rust and follows "zero-copy" principle. This allows it to make full use of the CPU by becoming cache and branch predictor friendly. Moreover, the architecture of ConnectorX ensures the data will be copied exactly once, directly from the source to the destination.

How does ConnectorX download the data?

Upon receiving the query, e.g. SELECT * FROM lineitem, ConnectorX will first issue a LIMIT 1 query SELECT * FROM lineitem LIMIT 1 to get the schema of the result set. Then, if partition_on is specified, ConnectorX will issue SELECT MIN(\$partition_on), MAX(\$partition_on) FROM (SELECT * FROM lineitem) to know the range of the partition column. After that, the original query is split into partitions based on the min/max information, e.g. SELECT * FROM (SELECT * FROM lineitem) WHERE \$partition_on > 0 AND \$partition_on < 10000. ConnectorX will then run a count query to get the partition size (e.g. SELECT COUNT(*) FROM (SELECT * FROM lineitem) WHERE \$partition_on > 0 AND \$partition_on < 10000). If the partition is not specified, the count query will be SELECT * FROM (SELECT * FROM lineitem). Finally, ConnectorX will use the schema info as well as the count info to allocate memory and download data by executing the queries normally.

Once the downloading begins, there will be one thread for each partition so that the data are downloaded in parallel at the partition level. The thread will issue the query of the corresponding partition to the database and then write the returned data to the destination row-wise or column-wise (depends on the database) in a streaming fashion.

This mechanism implies that having an index on the partition column is recommended to make full use of the parallel downloading power provided by ConnectorX.

Supported Sources & Destinations

Supported protocols, data types and type mappings can be found here.

Sources

  • <input type="checkbox" checked="" disabled="" /> Postgres
  • <input type="checkbox" checked="" disabled="" /> Mysql
  • <input type="checkbox" checked="" disabled="" /> Sqlite
  • <input type="checkbox" checked="" disabled="" /> Redshift (through postgres protocol)
  • <input type="checkbox" checked="" disabled="" /> Clickhouse (through mysql protocol)
  • <input type="checkbox" disabled="" /> SQL Server
  • <input type="checkbox" disabled="" /> Oracle
  • <input type="checkbox" disabled="" /> ...

Destinations

  • <input type="checkbox" checked="" disabled="" /> Pandas
  • <input type="checkbox" checked="" disabled="" /> PyArrow
  • <input type="checkbox" checked="" disabled="" /> Modin
  • <input type="checkbox" checked="" disabled="" /> Dask
  • <input type="checkbox" checked="" disabled="" /> Polars

Detailed Usage and Examples

API

connectorx.read_sql(conn: str, query: Union[List[str], str], *, return_type: str = "pandas", protocol: str = "binary", partition_on: Optional[str] = None, partition_range: Optional[Tuple[int, int]] = None, partition_num: Optional[int] = None)

Run the SQL query, download the data from database into a Pandas dataframe.

Parameters

  • conn: str: Connection string uri. Currently only PostgreSQL is supported.
  • query: Union[str, List[str]]: SQL query or list of SQL queries for fetching data.
  • return_type: str = "pandas": The return type of this function. It can be arrow, pandas, modin, dask or polars.
  • protocol: str = "binary": The protocol used to fetch data from source, default is binary. Check out here to see more details.
  • partition_on: Optional[str]: The column to partition the result.
  • partition_range: Optional[Tuple[int, int]]: The value range of the partition column.
  • partition_num: Optioinal[int]: The number of partitions to generate.

Examples

  • Read a DataFrame from a SQL using a single thread

    import connectorx as cx
    
    postgres_url = "postgresql://username:password@server:port/database"
    query = "SELECT * FROM lineitem"
    
    cx.read_sql(postgres_url, query)
    
  • Read a DataFrame parallelly using 10 threads by automatically partitioning the provided SQL on the partition column (partition_range will be automatically queried if not given)

    import connectorx as cx
    
    postgres_url = "postgresql://username:password@server:port/database"
    query = "SELECT * FROM lineitem"
    
    cx.read_sql(postgres_url, query, partition_on="l_orderkey", partition_num=10)
    
  • Read a DataFrame parallelly using 2 threads by manually providing two partition SQLs (the schemas of all the query results should be same)

    import connectorx as cx
    
    postgres_url = "postgresql://username:password@server:port/database"
    queries = ["SELECT * FROM lineitem WHERE l_orderkey <= 30000000", "SELECT * FROM lineitem WHERE l_orderkey > 30000000"]
    
    cx.read_sql(postgres_url, queries)
    
  • Read a DataFrame parallelly using 4 threads from a more complex query

    import connectorx as cx
    
    postgres_url = "postgresql://username:password@server:port/database"
    query = f"""
    SELECT l_orderkey,
           SUM(l_extendedprice * ( 1 - l_discount )) AS revenue,
           o_orderdate,
           o_shippriority
    FROM   customer,
           orders,
           lineitem
    WHERE  c_mktsegment = 'BUILDING'
           AND c_custkey = o_custkey
           AND l_orderkey = o_orderkey
           AND o_orderdate < DATE '1995-03-15'
           AND l_shipdate > DATE '1995-03-15'
    GROUP  BY l_orderkey,
              o_orderdate,
              o_shippriority 
    """
    
    cx.read_sql(postgres_url, query, partition_on="l_orderkey", partition_num=4)
    

Next Plan

Checkout our discussions to participate in deciding our next plan!

Historical Benchmark Results

https://sfu-db.github.io/connector-x/dev/bench/

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Files for connectorx, version 0.2.0
Filename, size File type Python version Upload date Hashes
Filename, size connectorx-0.2.0-cp37-cp37m-macosx_10_15_intel.whl (4.2 MB) File type Wheel Python version cp37 Upload date Hashes View
Filename, size connectorx-0.2.0-cp37-cp37m-manylinux2014_x86_64.whl (3.4 MB) File type Wheel Python version cp37 Upload date Hashes View
Filename, size connectorx-0.2.0-cp37-cp37m-win_amd64.whl (3.2 MB) File type Wheel Python version cp37 Upload date Hashes View
Filename, size connectorx-0.2.0-cp38-cp38-macosx_10_15_intel.whl (4.2 MB) File type Wheel Python version cp38 Upload date Hashes View
Filename, size connectorx-0.2.0-cp38-cp38-manylinux2014_x86_64.whl (3.4 MB) File type Wheel Python version cp38 Upload date Hashes View
Filename, size connectorx-0.2.0-cp38-cp38-win_amd64.whl (3.2 MB) File type Wheel Python version cp38 Upload date Hashes View
Filename, size connectorx-0.2.0-cp39-cp39-macosx_10_15_intel.whl (4.2 MB) File type Wheel Python version cp39 Upload date Hashes View
Filename, size connectorx-0.2.0-cp39-cp39-manylinux2014_x86_64.whl (3.4 MB) File type Wheel Python version cp39 Upload date Hashes View
Filename, size connectorx-0.2.0-cp39-cp39-win_amd64.whl (3.2 MB) File type Wheel Python version cp39 Upload date Hashes View

Supported by

AWS AWS Cloud computing Datadog Datadog Monitoring DigiCert DigiCert EV certificate Facebook / Instagram Facebook / Instagram PSF Sponsor Fastly Fastly CDN Google Google Object Storage and Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Salesforce Salesforce PSF Sponsor Sentry Sentry Error logging StatusPage StatusPage Status page