import pymongo
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq


# 連接 MongoDB
def connect_to_mongodb(host='localhost', port=27017, db_name='your_db_name', collection_name='your_collection_name'):
    client = pymongo.MongoClient(host, port)
    db = client[db_name]
    collection = db[collection_name]
    return collection


# 從 MongoDB 讀取數(shù)據(jù)
def read_from_mongodb(collection):
    data = list(collection.find())
    df = pd.DataFrame(data)
    return df


# 將 DataFrame 轉(zhuǎn)換為 Parquet 并保存
def convert_to_parquet(df, output_path='output.parquet'):
    table = pa.Table.from_pandas(df)
    pq.write_table(table, output_path)


if __name__ == "__main__":
    # 連接 MongoDB
    collection = connect_to_mongodb(db_name='test_db', collection_name='test_collection')
    # 從 MongoDB 讀取數(shù)據(jù)
    df = read_from_mongodb(collection)
    # 將數(shù)據(jù)存儲(chǔ)為 Parquet 格式
    convert_to_parquet(df, output_path='mongodb_data.parquet')

pip install pymongo pandas pyarrow

package main

import (
 "database/sql"
 "fmt"
 "log"

 _ "github.com/marcboeker/go-duckdb"
)

func main() {
    db := OpenDuckDB()
    rows, err := db.Query("SELECT id, first_name, family_name from read_parquet('employee.parquet');")
    if err != nil {
    	panic(fmt.Sprintf("failed to run select query against parquet file %v", err))
    }
    type Employee struct {
        Id         string `db:"id"`
        FirstName  string `db:"first_name"`
        FamilyName string `db:"family_name"`
    }
    
    var row Employee
    
    defer rows.Close()
    for rows.Next() {
        err := rows.Scan(&row.Id, &row.FirstName, &row.FamilyName)
        if err != nil {
            log.Fatal(err)
        }
        log.Printf("%v", row)
    }
    
    err = rows.Err()
    if err != nil {
    	log.Fatal(err)
    }
}

func OpenDuckDB() (db *sql.DB) {
    db, err := sql.Open("duckdb", "")
    if err != nil {
    	panic(fmt.Sprintf("failed to open parquet file using duckdb %v", err))
    }
    _, err = db.Exec("INSTALL parquet;")
    if err != nil {
    	panic(fmt.Sprintf("failed to INSTALL parquet extension. %v", err))
    }
    _, err = db.Exec("LOAD parquet;")
    if err != nil {
    	panic(fmt.Sprintf("failed to LOAD parquet extension. %v", err))
    }
    return
}