【发布时间】：2017-09-25 13:27:35
【问题描述】：
有人可以帮我理解我做错了什么吗？
以下所有内容均可按要求工作，但我遇到了可扩展性问题 -
On the first run i fetched ~70,000 rows into a blank table in ~2-3 s On the 2nd run i fetched ~80,000 rows into the same table in ~5 min On the 3rd run i fetched ~50,000 rows into the same table in ~30 min On the 4th run i fetched ~120,000 rows into the same table in ~1 hr On the 5th run i fetched ~100,000 rows into the same table in ~2 hr
每次我运行代码时，我都会看到客户端和数据库之间稳定的 ~600KB/s 流量，而此活动完成
如您所见，所有这些列的哈希检查似乎根本无法很好地扩展
我的代码试图完成什么？
我需要将每日库存数据添加到 postgres 数据库中。数据每天仅在源头更新一次，API 响应如下 -
{'instrument_token': '210011653'
'exchange_token': '820358'
'tradingsymbol': 'COLG17MAY1020.00PE'
'name': ''
'last_price': 0.0
'expiry': '2017-05-25'
'strike': 1020.0
'tick_size': 0.05
'lot_size': 700
'instrument_type': 'PE'
'segment': 'BFO-OPT'
'exchange': 'BFO'}
响应中的项目和行数每天都在变化 在给定的一天，我看到我可以在单个响应中获取 50,000 - 120,000 行（即大约 20-30 MB 的 csv 数据）。发送请求会获取给定日期的相同数据。
所以核心问题是 - 我想避免将同一行两次添加到数据库中，以防同一天多次获取数据。
到目前为止我做了什么 -
我是一个 db 新手，我的想法是自动增加一个 id 并添加一个 data_date 列，所以我的架构看起来像这样 -
CREATE TABLE IF NOT EXISTS instruments (
id bigserial,
data_date date NOT NULL,
instrument_token integer NOT NULL,
exchange_token integer NOT NULL,
tradingsymbol varchar(40) NOT NULL,
name varchar(40) NOT NULL,
last_price numeric(15,2) NOT NULL,
expiry date,
strike numeric(15,2),
tick_size numeric,
lot_size integer,
instrument_type varchar(10),
segment varchar(20),
exchange varchar(10),
PRIMARY KEY(id)
);
我已经建立了一个这样的类 -
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, mapper, relationship
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, Numeric, String, MetaData, Table, ForeignKey, DateTime, union
from sqlalchemy.engine.url import URL
engine = create_engine('postgresql://blah')
Base = declarative_base(engine)
def _unique(session, cls, hashfunc, queryfunc, constructor, arg, kw):
cache = getattr(session, '_unique_cache', None)
if cache is None:
session._unique_cache = cache = {}
key = (cls, hashfunc(*arg, **kw))
if key in cache:
return cache[key]
else:
with session.no_autoflush:
q = session.query(cls)
q = queryfunc(q, *arg, **kw)
obj = q.first()
if not obj:
obj = constructor(*arg, **kw)
session.add(obj)
cache[key] = obj
return obj
class UniqueMixin(object):
@classmethod
def unique_hash(cls, *arg, **kw):
raise NotImplementedError()
@classmethod
def unique_filter(cls, query, *arg, **kw):
raise NotImplementedError()
@classmethod
def as_unique(cls, session, *arg, **kw):
return _unique(
session,
cls,
cls.unique_hash,
cls.unique_filter,
cls,
arg, kw
)
class Instrument(UniqueMixin, Base):
__tablename__ = 'instruments'
__table_args__ = {'autoload':True}
__table__ = Table('instruments', Base.metadata,
Column('id', Integer, primary_key=True),
Column('data_date', String),
Column('instrument_token', Integer),
Column('exchange_token', Integer),
Column('tradingsymbol', String),
Column('name', String),
Column('last_price', Numeric),
Column('expiry', Integer),
Column('strike', Numeric),
Column('tick_size', Numeric),
Column('lot_size', Integer),
Column('instrument_type', String),
Column('segment', String),
Column('exchange', String))
@classmethod
def unique_hash(cls, data_date, instrument_token, exchange_token, tradingsymbol, name, last_price, expiry, strike, tick_size, lot_size, instrument_type, segment, exchange):
return data_date, instrument_token, exchange_token, tradingsymbol, name, last_price, expiry, strike, tick_size, lot_size, instrument_type, segment, exchange
@classmethod
def unique_filter(cls, query, data_date, instrument_token, exchange_token, tradingsymbol, name, last_price, expiry, strike, tick_size, lot_size, instrument_type, segment, exchange):
return query.filter(Instrument.data_date == data_date, Instrument.instrument_token == instrument_token, Instrument.exchange_token == exchange_token, Instrument.tradingsymbol == tradingsymbol, Instrument.name == name, Instrument.last_price == last_price, Instrument.expiry == expiry, Instrument.strike == strike, Instrument.tick_size == tick_size, Instrument.lot_size == lot_size, Instrument.instrument_type == instrument_type, Instrument.segment == segment, Instrument.exchange == exchange)
def __init__(self, data_date, instrument_token, exchange_token, tradingsymbol, name, last_price, expiry, strike, tick_size, lot_size, instrument_type, segment, exchange):
self.data_date = data_date
self.instrument_token = instrument_token
self.exchange_token = exchange_token
self.tradingsymbol = tradingsymbol
self.name = name
self.last_price = last_price
self.expiry = expiry
self.strike = strike
self.tick_size = tick_size
self.lot_size = lot_size
self.instrument_type = instrument_type
self.segment = segment
self.exchange = exchange
def __repr__(self):
return "<Instruments - '%s': '%s' - '%s' - '%s' - '%s' - '%s' - '%s' - '%s' - '%s' - '%s' - '%s' - '%s' - '%s' - '%s'>" % (
self.id,
self.data_date,
self.instrument_token,
self.exchange_token,
self.tradingsymbol,
self.name,
self.last_price,
self.expiry,
self.strike,
self.tick_size,
self.lot_size,
self.instrument_type,
self.segment,
self.exchange
)
插入数据的代码如下-
for instrument in response:
#print(instrument)
if instrument['expiry'] == '' :
instrument['expiry'] = null()
market_instrument = Instrument.as_unique(self.session,
data_date = datetime.date.today().isoformat(),
instrument_token = instrument['instrument_token'],
exchange_token = instrument['exchange_token'],
tradingsymbol = instrument['tradingsymbol'],
name = instrument['name'],
last_price = instrument['last_price'],
expiry = instrument['expiry'],
strike = instrument['strike'],
tick_size = instrument['tick_size'],
lot_size = instrument['lot_size'],
instrument_type = instrument['instrument_type'],
segment = instrument['segment'],
exchange = instrument['exchange'],
)
self.session.add(market_instrument)
self.session.commit()
我正在考虑的选项
你认为什么最好？
选项 1 不再使用 as_unique(
再创建一个 data_update_date 表 (data_date(primary), status(boolean))，该表在每日成功插入结束时更新
检查 data_update_date 获取今天的日期，如果存在则跳过整个块的添加
但是，这个选项并不能帮助我了解我的 as_unique 函数中是否还有其他需要纠正的错误
选项 2 使用 powa 和配置文件设置新数据库
查找并修复瓶颈
我正在使用官方的 postgres docker 映像，我遇到了用 hypopg 和其他必需的扩展扩展 debian 基础的死胡同
看起来centos会简单得多，所以我正在创建一个新的dockerfile来做到这一点
但是，由于我是 postgresql 和 sqlalchemy 的新手，我还需要您对我的代码是否存在一些明显问题提出意见
选项 3 只散列几列
我可以只散列前 3 个，不包括 id
但是我不知道该怎么做
只是减少hash classdef参数导致参数个数比类中定义的少，所以插入失败
选项 4
我没有与 postgresql 或 sqlalchemy 结婚
我应该改用非 ORM 方法吗？
或者，我应该使用数据库以外的东西来存储这种数据
我在 AWS 上的 m2.large 实例上运行它，它应该具有正确的性能，所以也许我使用了错误的方法来存储数据 如果在插入时出现这种情况，那么在进行技术分析时，多个线程将根本无法使用...
我应该改用 hadoop 之类的东西吗？
此外，此选项的一个明显缺点是另一个学习曲线可扩展为 hadoop...
【问题讨论】：
-
尝试分析并查看瓶颈在哪里。只是一个猜测：您的缓存变得如此之大，以至于它开始写入磁盘并从中读取。
-
plumSemPy 谢谢，是的，我正在整合 powa。我为此更新了问题中的选项 2
-
您需要让数据库处理唯一性（通过
UNIQUE约束或索引）。 -
univerio unique 无法按预期工作，因为它将应用于单个列
-
sqlalchemy 至少允许唯一的元组
标签： python database postgresql orm sqlalchemy