Pythonは本当に必要か？Rustで作るデータパイプライン

import csv
import random
import time

# Configuration
num_rows = 1_000_000
output_file = "large_data.csv"

# Define sample data
names = ["Alice", "Bob", "Charlie", "David", "Eve"]
countries = ["USA", "Canada", "UK", "Australia", "Germany", "Japan"]

# Generate CSV
start_time = time.time()
with open(output_file, mode="w", newline="") as file:
    writer = csv.writer(file)
    # Write header
    writer.writerow(["id", "name", "age", "country", "signup_date"])

    # Write rows
    for i in range(1, num_rows + 1):
        writer.writerow([
            i,  # id
            random.choice(names),  # name
            random.randint(18, 99),  # age
            random.choice(countries),  # country
            f"2024-{random.randint(1, 12):02d}-{random.randint(1, 28):02d}",  # signup_date
        ])

end_time = time.time()
print(f"Generated {num_rows} rows in {end_time - start_time:.2f} seconds. Output: {output_file}")

import asyncio
import concurrent.futures
import time

import pandas as pd
from google.cloud import bigquery


def insert_batch(batch, table_ref):
    client.insert_rows_json(table_ref, batch)

def insert_to_bigquery_python(csv_file, dataset_id, table_id):
    table_ref = client.dataset(dataset_id).table(table_id)

    start_time = time.time()

    df = pd.read_csv(csv_file)

    rows = df.to_dict(orient="records")

    async def process_batches():
        with concurrent.futures.ThreadPoolExecutor() as executor:
            loop = asyncio.get_event_loop()
            tasks = []
            for i in range(0, len(rows), 1000):
                batch = rows[i:i+1000]
                tasks.append(loop.run_in_executor(executor, insert_batch, batch, table_ref))
            await asyncio.gather(*tasks)

    asyncio.run(process_batches())

    print(f"Python Insert Time: {time.time() - start_time} seconds")

if __name__ == "__main__":
    client = bigquery.Client()
    insert_to_bigquery_python("large_data.csv", "DATASET_ID", "TABLE_ID")

use csv::Reader;
use serde_json::json;
use reqwest::Client;
use tokio::task;
use std::fs::File;
use std::io;
use std::time::Instant;
use std::env;

#[tokio::main]
async fn main() -> io::Result<()> {
    let csv_file = "large_data.csv";
    let dataset_id = "DATASET_ID";
    let table_id = "TABLE_ID";
    let bigquery_url = format!(
        "https://bigquery.googleapis.com/bigquery/v2/projects/{project_id}/datasets/{dataset_id}/tables/{table_id}/insertAll",
        project_id = "GCP_PROJECT_ID",
        dataset_id = dataset_id,
        table_id = table_id
    );

    let client = Client::new();
    let mut rdr = Reader::from_reader(File::open(csv_file)?);

    let headers = rdr.headers()?.clone(); // Clone headers to avoid borrowing `rdr`
    let mut rows = Vec::new();
    let start_time = Instant::now();

    let mut tasks = vec![];

    for result in rdr.records() {
        let record = result?;
        let mut json_record = json!({});
        for (header, value) in headers.iter().zip(record.iter()) {
            json_record[header] = json!(value);
        }
        rows.push(json_record);

        if rows.len() >= 1000 {
            let batch = rows.clone();
            rows.clear();
            let client = client.clone();
            let url = bigquery_url.clone();

            tasks.push(task::spawn(async move {
                insert_to_bigquery(&client, &url, &batch).await;
            }));
        }
    }

    if !rows.is_empty() {
        let batch = rows;
        let client = client.clone();
        let url = bigquery_url.clone();

        tasks.push(task::spawn(async move {
            insert_to_bigquery(&client, &url, &batch).await;
        }));
    }

    for task in tasks {
        task.await.unwrap();
    }

    println!("Rust Insert Time: {:?} seconds", start_time.elapsed().as_secs());
    Ok(())
}

async fn insert_to_bigquery(client: &Client, url: &str, rows: &[serde_json::Value]) {
    let access_token = env::var("GOOGLE_ACCESS_TOKEN").expect("GOOGLE_ACCESS_TOKEN environment variable is not set");

    let payload = json!({ "rows": rows.iter().map(|r| json!({ "json": r })).collect::<Vec<_>>() });

    let response = client.post(url)
        .header("Authorization", format!("Bearer {}", access_token))
        .json(&payload)
        .send()
        .await;

    match response {
        Ok(resp) if resp.status().is_success() => println!("Batch inserted successfully"),
        Ok(resp) => eprintln!("Failed to insert batch: {:?}", resp.text().await),
        Err(err) => eprintln!("Error sending request: {:?}", err),
    }
}