用C++实现高性能数据处理，Python实现操作Redshift导入数据文件。
在Visual Studio 2022中用C++和ODBC API导出SQL Server数据库中张表中的所有表的数据为CSV文件格式的数据流，用逗号作为分隔符，用双引号包裹每个数据，字符串类型的数据去掉前后的空格，数据中如果包含双引号，则将一个双引号替换为两个双引号，创建gzip压缩文件，输出数据流写入到gzip压缩文件包中的唯一的csv文件中，一张表对应一个gzip压缩文件。用异步编程的方法优化数据库表的读取和写入文件的过程，每个gzip压缩文件包中的csv文件的第一行都是用逗号分割的表头，对应数据库表的字段，程序需要异常处理和日志文件输出，当前路径下唯一的日志文件中包含运行时间戳、数据库名、schema名、表名、导出数据行数、成功或者失败的状态，日志文件不存在则创建。
以下是一个使用C++和ODBC API在Visual Studio 2022中实现将SQL Server数据库表数据导出为CSV文件并进行gzip压缩的示例代码。代码使用异步编程优化数据库读取和文件写入过程，并包含异常处理和日志记录功能。
#include <iostream>
#include <fstream>
#include <sstream>
#include <string>
#include <vector>
#include <future>
#include <sql.h>
#include <sqlext.h>
#include <zlib.h>
#include <ctime>
#include <iomanip>
// 日志记录函数
void logMessage(const std::string& message) {
std::ofstream logFile("export_log.txt", std::ios::app);
if (logFile.is_open()) {
auto now = std::chrono::system_clock::now();
std::time_t now_c = std::chrono::system_clock::to_time_t(now);
std::tm* now_tm = std::localtime(&now_c);
std::ostringstream oss;
oss << std::put_time(now_tm, "%Y-%m-%d %H:%M:%S") << " " << message << std::endl;
logFile << oss.str();
logFile.close();
}
}
// 处理字符串中的双引号
std::string escapeDoubleQuotes(const std::string& str) {
std::string result = str;
size_t pos = 0;
while ((pos = result.find('"', pos))!= std::string::npos) {
result.replace(pos, 1, 2, '"');
pos += 2;
}
return result;
}
// 从数据库读取表数据
std::vector<std::vector<std::string>> readTableData(SQLHSTMT hstmt) {
std::vector<std::vector<std::string>> data;
SQLSMALLINT columnCount = 0;
SQLNumResultCols(hstmt, &columnCount);
std::vector<SQLCHAR*> columns(columnCount);
std::vector<SQLINTEGER> lengths(columnCount);
for (SQLSMALLINT i = 0; i < columnCount; ++i) {
columns[i] = new SQLCHAR[SQL_MAX_MESSAGE_LENGTH];
SQLBindCol(hstmt, i + 1, SQL_C_CHAR, columns[i], SQL_MAX_MESSAGE_LENGTH, &lengths[i]);
}
while (SQLFetch(hstmt) == SQL_SUCCESS) {
std::vector<std::string> row;
for (SQLSMALLINT i = 0; i < columnCount; ++i) {
std::string value(reinterpret_cast<const char*>(columns[i]));
value = escapeDoubleQuotes(value);
row.push_back(value);
}
data.push_back(row);
}
for (SQLSMALLINT i = 0; i < columnCount; ++i) {
delete[] columns[i];
}
return data;
}
// 将数据写入CSV文件
void writeToCSV(const std::vector<std::vector<std::string>>& data, const std::vector<std::string>& headers, const std::string& filename) {
std::ofstream csvFile(filename);
if (csvFile.is_open()) {
// 写入表头
for (size_t i = 0; i < headers.size(); ++i) {
csvFile << '"' << headers[i] << '"';
if (i < headers.size() - 1) csvFile << ',';
}
csvFile << std::endl;
// 写入数据
for (const auto& row : data) {
for (size_t i = 0; i < row.size(); ++i) {
csvFile << '"' << row[i] << '"';
if (i < row.size() - 1) csvFile << ',';
}
csvFile << std::endl;
}
csvFile.close();
} else {
throw std::runtime_error("Failed to open CSV file for writing");
}
}
// 压缩CSV文件为gzip
void compressCSV(const std::string& csvFilename, const std::string& gzipFilename) {
std::ifstream csvFile(csvFilename, std::ios::binary);
std::ofstream gzipFile(gzipFilename, std::ios::binary);
if (csvFile.is_open() && gzipFile.is_open()) {
gzFile gzOut = gzopen(gzipFilename.c_str(), "wb");
if (gzOut) {
char buffer[1024];
while (csvFile.read(buffer, sizeof(buffer))) {
gzwrite(gzOut, buffer, sizeof(buffer));
}
gzwrite(gzOut, buffer, csvFile.gcount());
gzclose(gzOut);
} else {
throw std::runtime_error("Failed to open gzip file for writing");
}
csvFile.close();
gzipFile.close();
std::remove(csvFilename.c_str());
} else {
throw std::runtime_error("Failed to open files for compression");
}
}
// 导出单个表
void exportTable(const std::string& server, const std::string& database, const std::string& schema, const std::string& table) {
SQLHENV henv = nullptr;
SQLHDBC hdbc = nullptr;
SQLHSTMT hstmt = nullptr;
try {
SQLAllocHandle(SQL_HANDLE_ENV, SQL_NULL_HANDLE, &henv);
SQLSetEnvAttr(henv, SQL_ATTR_ODBC_VERSION, (SQLPOINTER)SQL_OV_ODBC3, 0);
SQLAllocHandle(SQL_HANDLE_DBC, henv, &hdbc);
std::string connectionString = "DRIVER={ODBC Driver 17 for SQL Server};SERVER=" + server + ";DATABASE=" + database + ";UID=your_username;PWD=your_password";
SQLRETURN ret = SQLDriverConnect(hdbc, nullptr, (SQLCHAR*)connectionString.c_str(), SQL_NTS, nullptr, 0, nullptr, SQL_DRIVER_NOPROMPT);
if (ret!= SQL_SUCCESS && ret!= SQL_SUCCESS_WITH_INFO) {
throw std::runtime_error("Failed to connect to database");
}
std::string query = "SELECT * FROM " + schema + "." + table;
SQLAllocHandle(SQL_HANDLE_STMT, hdbc, &hstmt);
ret = SQLExecDirect(hstmt, (SQLCHAR*)query.c_str(), SQL_NTS);
if (ret!= SQL_SUCCESS && ret!= SQL_SUCCESS_WITH_INFO) {
throw std::runtime_error("Failed to execute query");
}
std::vector<std::vector<std::string>> data = readTableData(hstmt);
std::vector<std::string> headers;
SQLSMALLINT columnCount = 0;
SQLNumResultCols(hstmt, &columnCount);
for (SQLSMALLINT i = 0; i < columnCount; ++i) {
SQLCHAR columnName[SQL_MAX_COLUMN_NAME_LEN];
SQLSMALLINT nameLen;
SQLDescribeCol(hstmt, i + 1, columnName, SQL_MAX_COLUMN_NAME_LEN, &nameLen, nullptr, nullptr, nullptr, nullptr);
headers.push_back(reinterpret_cast<const char*>(columnName));
}
std::string csvFilename = table + ".csv";
writeToCSV(data, headers, csvFilename);
std::string gzipFilename = table + ".gz";
compressCSV(csvFilename, gzipFilename);
std::ostringstream oss;
oss << "Database: " << database << ", Schema: " << schema << ", Table: " << table << ", Rows Exported: " << data.size() << ", Status: Success";
logMessage(oss.str());
} catch (const std::exception& e) {
std::ostringstream oss;
oss << "Database: " << database << ", Schema: " << schema << ", Table: " << table << ", Status: Failed, Error: " << e.what();
logMessage(oss.str());
} finally {
if (hstmt) SQLFreeHandle(SQL_HANDLE_STMT, hstmt);
if (hdbc) SQLDisconnect(hdbc); SQLFreeHandle(SQL_HANDLE_DBC, hdbc);
if (henv) SQLFreeHandle(SQL_HANDLE_ENV, henv);
}
}
int main() {
std::string server = "<实例名>";
std::string database = "<数据库名>";
std::string schema = "<Schema名>";
std::vector<std::string> tables = {"<表名1>", "<表名2>", "<表名3>"};
std::vector<std::future<void>> futures;
for (const auto& table : tables) {
futures.push_back(std::async(std::launch::async, exportTable, server, database, schema, table));
}
for (auto& fut : futures) {
fut.get();
}
return 0;
}
代码说明：
日志记录： logMessage 函数用于记录操作日志，包括时间戳、数据库名、schema名、表名、导出数据行数和操作状态。
字符串处理： escapeDoubleQuotes 函数用于处理字符串中的双引号，将其替换为两个双引号。
数据库读取： readTableData 函数使用ODBC API从数据库中读取表数据，并将其存储在二维向量中。
CSV写入： writeToCSV 函数将数据写入CSV文件，包括表头和数据行，并用双引号包裹每个数据，使用逗号作为分隔符。
文件压缩： compressCSV 函数将生成的CSV文件压缩为gzip格式，并删除原始CSV文件。
表导出： exportTable 函数负责连接数据库、执行查询、读取数据、写入CSV文件并压缩。
主函数： main 函数定义了数据库服务器、数据库名、schema名和表名，并使用异步任务并行导出每个表的数据。
用Python删除当前目录下所有功能扩展名为gz文件，接着运行export_sqlserver.exe程序，输出该程序的输出内容并等待它运行完成，然后连接SQL Server数据库和Amazon Redshift数据仓库，从数据库中获取所有表和它们的字段名，然后在Redshift中创建字段名全部相同的同名表，字段长度全部为最长的varchar类型，如果表已经存在则不创建表，自动上传当前目录下所有功能扩展名为gz文件到S3，默认覆盖同名的文件，然后使用COPY INTO将S3上包含csv文件的gz压缩包导入对应创建的Redshift表中，文件数据的第一行是表头，导入所有上传的文件到Redshift表，程序需要异常处理和日志文件输出，当前路径下唯一的日志文件中包含运行时间戳、数据库名、schema名、表名、导入数据行数、成功或者失败的状态，日志文件不存在则创建。
import os
import subprocess
import pyodbc
import redshift_connector
import boto3
import logging
from datetime import datetime
# 配置日志记录
logging.basicConfig(filename='operation_log.log', level=logging.INFO,
format='%(asctime)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S')
def delete_gz_files():
try:
for file in os.listdir('.'):
if file.endswith('.gz'):
os.remove(file)
logging.info('所有.gz文件已删除')
except Exception as e:
logging.error(f'删除.gz文件时出错: {e}')
def run_export_sqlserver():
try:
result = subprocess.run(['export_sqlserver.exe'], capture_output=True, text=True)
print(result.stdout)
logging.info('export_sqlserver.exe运行成功')
except Exception as e:
logging.error(f'运行export_sqlserver.exe时出错: {e}')
def create_redshift_tables():
# SQL Server 连接配置
sqlserver_conn_str = 'DRIVER={ODBC Driver 17 for SQL Server};SERVER=your_sqlserver_server;DATABASE=your_database;UID=your_username;PWD=your_password'
try:
sqlserver_conn = pyodbc.connect(sqlserver_conn_str)
sqlserver_cursor = sqlserver_conn.cursor()
# Redshift 连接配置
redshift_conn = redshift_connector.connect(
host='your_redshift_host',
database='your_redshift_database',
user='your_redshift_user',
password='your_redshift_password',
port=5439
)
redshift_cursor = redshift_conn.cursor()
sqlserver_cursor.execute("SELECT TABLE_NAME FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_TYPE = 'BASE TABLE'")
tables = sqlserver_cursor.fetchall()
for table in tables:
table_name = table[0]
sqlserver_cursor.execute(f"SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME = '{table_name}'")
columns = sqlserver_cursor.fetchall()
column_definitions = ', '.join([f"{column[0]} VARCHAR(MAX)" for column in columns])
try:
redshift_cursor.execute(f"CREATE TABLE IF NOT EXISTS {table_name} ({column_definitions})")
redshift_conn.commit()
logging.info(f'在Redshift中成功创建表 {table_name}')
except Exception as e:
logging.error(f'在Redshift中创建表 {table_name} 时出错: {e}')
sqlserver_conn.close()
redshift_conn.close()
except Exception as e:
logging.error(f'连接数据库或创建表时出错: {e}')
def upload_gz_files_to_s3():
s3 = boto3.client('s3')
bucket_name = 'your_bucket_name'
try:
for file in os.listdir('.'):
if file.endswith('.gz'):
s3.upload_file(file, bucket_name, file)
logging.info(f'成功上传文件 {file} 到S3')
except Exception as e:
logging.error(f'上传文件到S3时出错: {e}')
def copy_data_to_redshift():
redshift_conn = redshift_connector.connect(
host='your_redshift_host',
database='your_redshift_database',
user='your_redshift_user',
password='your_redshift_password',
port=5439
)
redshift_cursor = redshift_conn.cursor()
bucket_name = 'your_bucket_name'
try:
for file in os.listdir('.'):
if file.endswith('.gz') and file.endswith('.csv.gz'):
table_name = file.split('.')[0]
s3_path = f's3://{bucket_name}/{file}'
sql = f"COPY {table_name} FROM '{s3_path}' IAM_ROLE 'your_iam_role' CSV HEADER"
try:
redshift_cursor.execute(sql)
redshift_conn.commit()
row_count = redshift_cursor.rowcount
logging.info(f'成功将数据导入表 {table_name}，导入行数: {row_count}')
except Exception as e:
logging.error(f'将数据导入表 {table_name} 时出错: {e}')
except Exception as e:
logging.error(f'连接Redshift或导入数据时出错: {e}')
finally:
redshift_conn.close()
if __name__ == "__main__":
delete_gz_files()
run_export_sqlserver()
create_redshift_tables()
upload_gz_files_to_s3()
copy_data_to_redshift()
代码说明：
日志记录：使用 logging 模块配置日志记录，记录操作的时间戳和操作信息到 operation_log.log 文件。
删除.gz文件： delete_gz_files 函数删除当前目录下所有扩展名为 .gz 的文件。
运行export_sqlserver.exe： run_export_sqlserver 函数运行 export_sqlserver.exe 程序并输出其内容。
创建Redshift表： create_redshift_tables 函数连接SQL Server和Redshift数据库，获取SQL Server中所有表和字段名，在Redshift中创建同名表，字段类型为 VARCHAR(MAX) 。
上传.gz文件到S3： upload_gz_files_to_s3 函数上传当前目录下所有扩展名为 .gz 的文件到S3。
将数据从S3导入Redshift： copy_data_to_redshift 函数使用 COPY INTO 语句将S3上的CSV压缩包数据导入对应的Redshift表中。
请根据实际的数据库配置、S3桶名和IAM角色等信息修改代码中的相关参数。
![P1044 [NOIP2003 普及组] 栈 C语言](https://i-blog.csdnimg.cn/direct/d911dcb464b64c27a836bb5b3d78dde8.png)
