如何通过Polars从MySQL数据库读取DataFrame

Great question! Right now, Polars doesn’t have built-in support for reading directly from MySQL or other SQL databases. That means you’ll need to pair it with a database driver (like sqlx or the mysql crate) to fetch your data first, then convert it into a Polars DataFrame. Your current approach (fetching into a Vec<Struct> then transforming to Series) is totally valid—let’s break down how to refine that, plus share a couple of cleaner implementations.

Option 1: Using sqlx (Asynchronous) #

sqlx is a popular async SQL toolkit that works seamlessly with Rust’s async runtimes like Tokio. Here’s a step-by-step implementation:

Step 1: Add Dependencies #

First, update your Cargo.toml with the required crates:

[dependencies]
sqlx = { version = "0.7", features = ["mysql", "runtime-tokio-native-tls"] }
polars = { version = "0.32", features = ["lazy", "serde"] }
tokio = { version = "1.0", features = ["full"] }

Step 2: Define Your Data Struct #

Create a struct that matches your MySQL table schema, and implement sqlx::FromRow (this lets sqlx automatically map query results to your struct):

use sqlx::FromRow;
use polars::prelude::*;

#[derive(Debug, FromRow)]
struct User {
    id: i32,
    name: String,
    email: String,
    age: Option<u8>, // Use Option for nullable columns
}

Step 3: Fetch and Convert to Polars DataFrame #

Connect to your database, fetch the data into a Vec<User>, then extract each field into a vector to build Polars Series:

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Create a connection pool to your MySQL database
    let pool = sqlx::mysql::MySqlPoolOptions::new()
        .max_connections(5)
        .connect("mysql://your_username:your_password@localhost/your_database")
        .await?;

    // Fetch all users into a Vec<User>
    let users = sqlx::query_as!(User, "SELECT id, name, email, age FROM users")
        .fetch_all(&pool)
        .await?;

    // Extract each column's values into separate vectors
    let ids: Vec<i32> = users.iter().map(|user| user.id).collect();
    let names: Vec<String> = users.iter().map(|user| user.name.clone()).collect();
    let emails: Vec<String> = users.iter().map(|user| user.email.clone()).collect();
    let ages: Vec<Option<u8>> = users.iter().map(|user| user.age).collect();

    // Create Polars Series for each column
    let s_id = Series::new("id", ids);
    let s_name = Series::new("name", names);
    let s_email = Series::new("email", emails);
    let s_age = Series::new("age", ages);

    // Build the final DataFrame
    let df = DataFrame::new(vec![s_id, s_name, s_email, s_age])?;

    // Print the result to verify
    println!("{}", df);

    Ok(())
}

Option 2: Using the mysql Crate (Synchronous) #

If you don’t need async functionality, the mysql crate offers a straightforward synchronous API:

Step 1: Add Dependencies #

Update Cargo.toml:

[dependencies]
mysql = "23.0"
polars = { version = "0.32", features = ["lazy"] }

Step 2: Define Struct and Implement FromRow #

use mysql::prelude::*;
use polars::prelude::*;

#[derive(Debug, PartialEq, Eq)]
struct User {
    id: i32,
    name: String,
    email: String,
    age: Option<u8>,
}

// Implement FromRow to map MySQL rows to your struct
impl FromRow for User {
    fn from_row(row: mysql::Row) -> Self {
        User {
            id: row.get(0).unwrap(),
            name: row.get(1).unwrap(),
            email: row.get(2).unwrap(),
            age: row.get(3).unwrap(),
        }
    }
}

Step 3: Fetch and Convert #

fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Connect to the database
    let db_url = "mysql://your_username:your_password@localhost/your_database";
    let pool = mysql::Pool::new(db_url)?;
    let mut conn = pool.get_conn()?;

    // Fetch data into Vec<User>
    let users: Vec<User> = conn.query_map(
        "SELECT id, name, email, age FROM users",
        |(id, name, email, age)| User { id, name, email, age },
    )?;

    // Convert to Polars DataFrame (same as the async example)
    let s_id = Series::new("id", users.iter().map(|u| u.id).collect::<Vec<_>>());
    let s_name = Series::new("name", users.iter().map(|u| u.name.clone()).collect::<Vec<_>>());
    let s_email = Series::new("email", users.iter().map(|u| u.email.clone()).collect::<Vec<_>>());
    let s_age = Series::new("age", users.iter().map(|u| u.age).collect::<Vec<_>>());

    let df = DataFrame::new(vec![s_id, s_name, s_email, s_age])?;
    println!("{}", df);

    Ok(())
}

Refining Your Original Approach #

Your initial fold method works, but using iter().map().collect() makes the code more readable and concise. For example, your sample code can be rewritten as:

struct A(u8, i8);
fn main() {
    let v = vec![A(1, 4), A(2, 6), A(3, 5)];
    
    // Extract each field into separate vectors
    let unsigneds: Vec<u8> = v.iter().map(|a| a.0).collect();
    let signeds: Vec<i8> = v.iter().map(|a| a.1).collect();
    
    // Build Series and DataFrame
    let s0 = Series::new("unsigned", unsigneds);
    let s1 = Series::new("signed", signeds);
    
    let df = DataFrame::new(vec![s0, s1]).unwrap();
    println!("{}", df);
}

Key Notes #

• Performance: Manual extraction of vectors (like we did above) is the most efficient way to convert to Polars Series, as it avoids any extra serialization overhead.
• Large Datasets: For very large datasets, consider streaming results instead of loading everything into a Vec first (both sqlx and mysql support streaming rows). You could build Series incrementally or use Polars’ lazy API to process data in chunks.
• Automation: If you have a struct with many fields, you could use a macro or code generation tool to automate the vector extraction step, but this adds complexity—weigh it against your project’s needs.

内容的提问来源于stack exchange，提问作者Hamza Zubair