arrow-bridge 0.2.0

Arrow-based pipeline accelerators for databases and cloud storage

Arrow Bridge

Arrow Bridge (arrow_bridge) provides pipeline accelerators for working with databases and cloud storage using PyArrow. It helps you extract, transform, and load data efficiently from relational databases and cloud storage (AWS S3, Azure Blob, GCS) into PyArrow Tables and multiple file formats.

Features

• Database connectors: PostgreSQL, MySQL, SQL Server (via SQLAlchemy)
• Database extractors: Extract tables in chunks into PyArrow Tables
• Cloud storage connectors: AWS S3 (boto3), Azure Blob Storage, GCS
• Cloud storage extractors: Extract files (CSV, Parquet, ORC, JSON, XML, TXT) into PyArrow Tables
• PyArrow Table writer: Write PyArrow Tables to local or cloud storage in multiple formats (Parquet, Feather, CSV, JSON, ORC)

Installation

pip install arrow-bridge

"""
Example main pipeline using arrow-bridge abstractions.

Scenarios:
1. Extract from PostgreSQL → Parquet → AWS S3
2. Extract from SQL Server → ORC → Azure Blob
3. Extract from S3 → Avro → S3
"""

from arrow_bridge.databases.connector import connect_postgres, connect_sqlserver
from arrow_bridge.databases.extractor import extract_table_to_arrow
from arrow_bridge.storage.connector import connect_s3, connect_azure_blob
from arrow_bridge.storage.extractor import extract_s3_file_to_arrow
from arrow_bridge.arrow_writer import write_arrow_table

# ---------------------------
# Scenario 1: Postgres → Parquet → S3
# ---------------------------
# Connect to Postgres
pg_engine = connect_postgres(
    user="postgres_user",
    password="postgres_pass",
    host="postgres_host",
    port=5432,
    database="my_database"
)

# Extract table to Arrow
arrow_table_pg = extract_table_to_arrow(
    engine=pg_engine,
    schema="public",
    table="sales",
    chunk_size=5000
)

# Connect to S3
s3_client = connect_s3(
    access_key="AWS_ACCESS_KEY",
    secret_key="AWS_SECRET_KEY",
    region_name="us-east-1"
)

# Write to S3 in Parquet
write_arrow_table(
    table=arrow_table_pg,
    destination_path="output/sales_pg.parquet",
    file_format="parquet",
    storage_type="s3",
    s3_client=s3_client,
    container_or_bucket="my-bucket"
)

# ---------------------------
# Scenario 2: SQL Server → ORC → Azure Blob
# ---------------------------
# Connect to SQL Server
sqlsrv_engine = connect_sqlserver(
    user="sql_user",
    password="sql_pass",
    host="sql_server_host",
    port=1433,
    database="my_database"
)

# Extract table to Arrow
arrow_table_sqlsrv = extract_table_to_arrow(
    engine=sqlsrv_engine,
    schema="dbo",
    table="customers",
    chunk_size=5000
)

# Connect to Azure Blob
azure_client = connect_azure_blob(
    connection_string=(
        "DefaultEndpointsProtocol=https;"
        "AccountName=...;"
        "AccountKey=...;"
        "EndpointSuffix=core.windows.net"
    )
)

# Write to Azure Blob in ORC
write_arrow_table(
    table=arrow_table_sqlsrv,
    destination_path="output/customers_orc.orc",
    file_format="orc",
    storage_type="azure",
    azure_blob_client=azure_client,
    container_or_bucket="my-container"
)