python连接pg数据库做数据分析

### 使用 Python 连接 PostgreSQL 数据库 为了实现 Python 和 PostgreSQL 的连接，通常会使用 `psycopg2` 库。此库允许开发者通过编写简单的 Python 脚本来执行复杂的数据库操作。 建立到 PostgreSQL 数据库的连接涉及几个重要参数：数据库名称 (`dbname`)、用户名 (`user`)、密码 (`password`) 及端口号 (`port`)。下面是一个具体的例子来展示如何创建这种连接： ```python import psycopg2 conn = psycopg2.connect( dbname='goog_stock', user='postgres', password='LlQ54951', port='5432' ) cursor = conn.cursor() ``` 上述代码片段展示了怎样配置并初始化一个指向特定 PostgreSQL 数据库实例的新连接对象[^1]。一旦建立了连接，就可以利用游标对象 (`cursor`) 来发送 SQL 查询给服务器以及处理返回的数据集。 对于更复杂的应用场景或需要频繁访问数据库的情况，建议考虑将这些设置封装成函数或是类的一部分以便重用和维护。 #### 执行查询语句 当成功建立起与数据库之间的桥梁之后，便可以构建 SQL 查询字符串并通过游标的 execute 方法传递它。例如读取指定日期范围内的股票交易记录如下所示: ```sql SQL = ''' SELECT dt, high, low, open, close, volume, adj_close FROM "GOOG" WHERE dt BETWEEN '2016-11-08' AND '2016-11-09' ORDER BY dt LIMIT 100; ''' ``` 接着调用 pandas 的 read_sql 函数可以直接获取结构化的 DataFrame 表格形式的结果集，这使得后续数据分析变得更加直观方便: ```python import pandas as pd df = pd.read_sql(SQL, conn, parse_dates=['dt']) print(df.head()) ``` 最后，在完成所有的数据库交互活动后记得关闭游标及断开连接以释放资源： ```python cursor.close() conn.close() ```

<think>我们需要通过ZooKeeper获取PostgreSQL的连接配置，然后使用psycopg2连接数据库并执行SQL查询。 步骤： 1. 连接ZooKeeper，从指定节点获取PostgreSQL的连接配置（如host, port, dbname, user, password等）。 2. 解析配置（可能是JSON格式或其他格式）。 3. 使用psycopg2建立数据库连接。 4. 执行SQL查询。 5. 关闭数据库连接和ZooKeeper连接。 需要安装的库： - kazoo：用于连接ZooKeeper - psycopg2：用于连接PostgreSQL 安装命令： pip install kazoo psycopg2-binary 注意：这里使用psycopg2-binary简化安装。 实现步骤： 1. 连接ZooKeeper，获取配置数据。 2. 将获取的配置数据（假设是JSON字符串）解析为字典。 3. 使用解析后的配置创建PostgreSQL连接。 4. 执行SQL查询。 示例代码： 假设ZooKeeper中节点路径为：/config/db/postgresql，节点数据为JSON字符串，例如： {"host": "localhost", "port": 5432, "dbname": "testdb", "user": "postgres", "password": "secret"} 代码结构： - 创建ZooKeeper客户端，连接ZooKeeper服务器。 - 从指定节点获取数据。 - 解析数据。 - 使用psycopg2连接数据库并执行查询。 注意：ZooKeeper连接是异步的，我们这里使用同步方式（简化示例）。 完整示例：</think># 使用 ZooKeeper 配置连接 PostgreSQL 并执行 SQL 查询 在分布式系统中，通过 ZooKeeper 管理数据库连接配置是一种高效的方式，可以实现配置的集中管理和动态更新。下面我将详细介绍如何在 Python 中实现这一过程。 ## 整体架构 ``` +----------------+ +----------------+ +----------------+ | Python应用 | ---> | ZooKeeper | ---> | PostgreSQL | | (执行SQL查询) | <--- | (存储配置信息) | <--- | (数据库服务) | +----------------+ +----------------+ +----------------+ ``` ## 实现步骤 ### 1. 安装所需库 ```bash pip install kazoo psycopg2-binary configparser ``` ### 2. ZooKeeper 配置管理类 ```python from kazoo.client import KazooClient import json import logging class ZooKeeperConfigManager: def __init__(self, zk_hosts='localhost:2181', config_path='/db_config/postgresql'): self.zk_hosts = zk_hosts self.config_path = config_path self.zk = KazooClient(hosts=zk_hosts) self.zk.start() logging.basicConfig(level=logging.INFO) def get_db_config(self): """从ZooKeeper获取数据库配置""" if self.zk.exists(self.config_path): data, _ = self.zk.get(self.config_path) return json.loads(data.decode('utf-8')) else: raise Exception(f"配置路径 {self.config_path} 不存在") def watch_config(self, callback): """监听配置变更""" @self.zk.DataWatch(self.config_path) def data_watch(data, stat, event): if event and event.type == "CHANGED": logging.info("检测到数据库配置变更") new_config = json.loads(data.decode('utf-8')) callback(new_config) def close(self): self.zk.stop() self.zk.close() ``` ### 3. PostgreSQL 数据库操作类 ```python import psycopg2 from psycopg2 import pool class PostgreSQLManager: def __init__(self, config): self.config = config self.connection_pool = self.create_connection_pool() def create_connection_pool(self): """创建连接池""" return psycopg2.pool.SimpleConnectionPool( minconn=1, maxconn=10, host=self.config['host'], port=self.config['port'], database=self.config['database'], user=self.config['user'], password=self.config['password'] ) def execute_query(self, sql, params=None): """执行查询并返回结果""" conn = self.connection_pool.getconn() try: with conn.cursor() as cursor: cursor.execute(sql, params or ()) result = cursor.fetchall() return result finally: self.connection_pool.putconn(conn) def execute_update(self, sql, params=None): """执行更新操作""" conn = self.connection_pool.getconn() try: with conn.cursor() as cursor: cursor.execute(sql, params or ()) conn.commit() return cursor.rowcount except Exception as e: conn.rollback() raise e finally: self.connection_pool.putconn(conn) def update_config(self, new_config): """更新数据库配置（当ZooKeeper配置变更时调用）""" self.config = new_config self.connection_pool.closeall() self.connection_pool = self.create_connection_pool() def close(self): self.connection_pool.closeall() ``` ### 4. 整合使用 ```python def main(): # 初始化ZooKeeper配置管理器 zk_manager = ZooKeeperConfigManager( zk_hosts='zk1:2181,zk2:2181,zk3:2181', config_path='/services/db/postgresql' ) try: # 获取初始配置 db_config = zk_manager.get_db_config() print("获取数据库配置:", db_config) # 初始化数据库管理器 db_manager = PostgreSQLManager(db_config) # 设置配置变更监听 def config_change_callback(new_config): print("接收到新配置:", new_config) db_manager.update_config(new_config) zk_manager.watch_config(config_change_callback) # 示例：执行SQL查询 users = db_manager.execute_query("SELECT * FROM users WHERE age > %s", (25,)) print("查询结果:", users) # 示例：执行更新操作 update_count = db_manager.execute_update( "UPDATE users SET status = %s WHERE last_login < NOW() - INTERVAL '6 months'", ('inactive',) ) print(f"更新了 {update_count} 条记录") # 保持程序运行（在实际应用中可能是Web服务器或长期运行的服务） import time while True: time.sleep(10) finally: db_manager.close() zk_manager.close() if __name__ == "__main__": main() ``` ## 关键组件说明 ### 1. ZooKeeper 配置节点结构 在 ZooKeeper 中创建 `/services/db/postgresql` 节点，数据格式为 JSON： ```json { "host": "db-host.example.com", "port": 5432, "database": "mydb", "user": "app_user", "password": "secure_password" } ``` ### 2. 配置更新流程 ``` 1. 管理员更新ZooKeeper中的配置 2. ZooKeeper通知所有监听该节点的客户端 3. 客户端回调函数被触发 4. 数据库连接池使用新配置重建 5. 后续查询使用新的数据库连接 ``` ### 3. 连接池的优势 使用连接池（`SimpleConnectionPool`）可以： 1. 减少连接创建/销毁的开销 2. 控制最大连接数防止过载 3. 提高高并发场景下的性能 4. 支持平滑配置切换 ## 最佳实践 1. **安全存储密码**： ```python # 使用环境变量或专用秘钥管理服务 import os password = os.getenv('DB_PASSWORD') ``` 2. **错误处理与重试**： ```python from kazoo.retry import KazooRetry retry_policy = KazooRetry(max_tries=3, delay=0.5) zk = KazooClient(hosts=zk_hosts, connection_retry=retry_policy) ``` 3. **连接验证**： ```python # 添加连接验证方法 def verify_connection(self): conn = self.connection_pool.getconn() try: conn.cursor().execute("SELECT 1") return True except: return False finally: self.connection_pool.putconn(conn) ``` 4. **配置版本控制**： ```python # 在配置中添加版本号 config = { "version": "2023-07-01", "host": "db-host.example.com", # ...其他配置 } ``` ## 性能优化 1. **连接池大小**：根据应用负载动态调整 $$ \text{最佳连接数} = \frac{\text{核心数} \times 2}{\text{平均查询时间}} $$ 2. **批量操作**：减少数据库交互次数 ```python # 批量插入示例 with conn.cursor() as cursor: cursor.executemany( "INSERT INTO logs (timestamp, message) VALUES (%s, %s)", log_entries ) ``` 3. **查询优化**：使用 EXPLAIN 分析慢查询 ## 典型应用场景 1. 微服务架构中的数据库配置管理 2. 多环境（开发/测试/生产）统一配置 3. 数据库故障转移和主从切换 4. 动态扩缩容数据库集群 ## 注意事项 1. **ZooKeeper 可用性**：确保 ZooKeeper 集群的高可用 2. **配置回退机制**：当新配置无效时能回退到旧配置 3. **连接泄露检查**：定期验证连接池状态 4. **敏感信息加密**：对密码等敏感信息进行加密存储 通过这种架构，您的应用可以实时响应数据库配置变更，无需重启服务即可切换到新的数据库连接设置，大大提高了系统的灵活性和可靠性。 ---