在 SQLAlchemy 中映射许多类似的表

Question

我有许多（约 2000 个）带有时间序列数据的位置。每个时间序列都有数百万行。我想将这些存储在 Postgres 数据库中。我目前的方法是为每个位置时间序列创建一个表，以及一个存储每个位置信息（坐标、海拔等）的元表。我正在使用 Python/SQLAlchemy 创建和填充表。我想在元表和每个时间序列表之间建立关系，以执行诸如“选择在日期 A 和日期 B 之间具有数据的所有位置”和“选择日期 A 的所有数据并导出带有坐标的 csv”之类的查询。创建多个具有相同结构（只是名称不同）并与元表有关系的表的最佳方法是什么？还是应该使用不同的数据库设计？

目前我正在使用这种方法来生成很多类似的映射：

from sqlalchemy import create_engine, MetaData
from sqlalchemy.types import Float, String, DateTime, Integer
from sqlalchemy import Column, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship, backref

Base = declarative_base()


def make_timeseries(name):
    class TimeSeries(Base):

        __tablename__ = name
        table_name = Column(String(50), ForeignKey('locations.table_name'))
        datetime = Column(DateTime, primary_key=True)
        value = Column(Float)

        location = relationship('Location', backref=backref('timeseries',
                                lazy='dynamic'))

        def __init__(self, table_name, datetime, value):
            self.table_name = table_name
            self.datetime = datetime
            self.value = value

        def __repr__(self):
            return "{}: {}".format(self.datetime, self.value)

    return TimeSeries


class Location(Base):

    __tablename__ = 'locations'
    id = Column(Integer, primary_key=True)
    table_name = Column(String(50), unique=True)
    lon = Column(Float)
    lat = Column(Float)

if __name__ == '__main__':
    connection_string = 'postgresql://user:pw@localhost/location_test'
    engine = create_engine(connection_string)
    metadata = MetaData(bind=engine)
    Session = sessionmaker(bind=engine)
    session = Session()

    TS1 = make_timeseries('ts1')
    # TS2 = make_timeseries('ts2')   # this breaks because of the foreign key
    Base.metadata.create_all(engine)
    session.add(TS1("ts1", "2001-01-01", 999))
    session.add(TS1("ts1", "2001-01-02", -555))

    qs = session.query(Location).first()
    print qs.timeseries.all()

这种方法有一些问题，最值得注意的是，如果我创建多个TimeSeries，则外键不起作用。以前我使用过一些变通方法，但这一切似乎都是一个大技巧，我觉得必须有更好的方法来做到这一点。我应该如何组织和访问我的数据？

Answer 1

Alternative-1: Table Partitioning

Partitioning 一读到完全相同的表结构就会立即浮现在脑海中。我不是 DBA，也没有太多使用它的生产经验（在 PostgreSQL 上更是如此），但是 请阅读PostgreSQL - Partitioning 文档。表分区旨在准确解决您遇到的问题，但是超过 1K 的表/分区听起来很有挑战性；因此，请在论坛/SO 上进行更多研究，以了解有关此主题的可扩展性相关问题。

鉴于您最常用的两个搜索条件，datetime 组件非常重要，因此必须有可靠的索引策略。如果您决定使用 partitioning 根目录，则明显的分区策略将基于日期范围。与最新数据相比，这可能允许您将旧数据划分为不同的块，特别是假设旧数据（几乎从未）更新，因此物理布局将是密集且高效的；而您可以采用另一种策略来获取更多“最新”数据。

Alternative-2: trick SQLAlchemy

这基本上使您的示例代码工作通过欺骗 SA 假设所有那些 TimeSeries 是使用 Concrete Table Inheritance 的一个实体的 children。下面的代码是自包含的，它创建了 50 个表，其中包含最少的数据。但是如果你已经有一个数据库，它应该可以让你相当快地检查性能，这样你就可以做出决定，如果它是很可能的。

from datetime import date, datetime

from sqlalchemy import create_engine, Column, String, Integer, DateTime, Float, ForeignKey, func
from sqlalchemy.orm import sessionmaker, relationship, configure_mappers, joinedload
from sqlalchemy.ext.declarative import declarative_base, declared_attr
from sqlalchemy.ext.declarative import AbstractConcreteBase, ConcreteBase


engine = create_engine('sqlite:///:memory:', echo=True)
Session = sessionmaker(bind=engine)
session = Session()
Base = declarative_base(engine)


# MODEL
class Location(Base):
    __tablename__ = 'locations'
    id = Column(Integer, primary_key=True)
    table_name = Column(String(50), unique=True)
    lon = Column(Float)
    lat = Column(Float)


class TSBase(AbstractConcreteBase, Base):
    @declared_attr
    def table_name(cls):
        return Column(String(50), ForeignKey('locations.table_name'))


def make_timeseries(name):
    class TimeSeries(TSBase):
        __tablename__ = name
        __mapper_args__ = { 'polymorphic_identity': name, 'concrete':True}

        datetime = Column(DateTime, primary_key=True)
        value = Column(Float)

        def __init__(self, datetime, value, table_name=name ):
            self.table_name = table_name
            self.datetime = datetime
            self.value = value

    return TimeSeries


def _test_model():
    _NUM = 50
    # 0. generate classes for all tables
    TS_list = [make_timeseries('ts{}'.format(1+i)) for i in range(_NUM)]
    TS1, TS2, TS3 = TS_list[:3] # just to have some named ones
    Base.metadata.create_all()
    print('-'*80)

    # 1. configure mappers
    configure_mappers()

    # 2. define relationship
    Location.timeseries = relationship(TSBase, lazy="dynamic")
    print('-'*80)

    # 3. add some test data
    session.add_all([Location(table_name='ts{}'.format(1+i), lat=5+i, lon=1+i*2)
        for i in range(_NUM)])
    session.commit()
    print('-'*80)

    session.add(TS1(datetime(2001,1,1,3), 999))
    session.add(TS1(datetime(2001,1,2,2), 1))
    session.add(TS2(datetime(2001,1,2,8), 33))
    session.add(TS2(datetime(2002,1,2,18,50), -555))
    session.add(TS3(datetime(2005,1,3,3,33), 8))
    session.commit()


    # Query-1: get all timeseries of one Location
    #qs = session.query(Location).first()
    qs = session.query(Location).filter(Location.table_name == "ts1").first()
    print(qs)
    print(qs.timeseries.all())
    assert 2 == len(qs.timeseries.all())
    print('-'*80)


    # Query-2: select all location with data between date-A and date-B
    dateA, dateB = date(2001,1,1), date(2003,12,31)
    qs = (session.query(Location)
            .join(TSBase, Location.timeseries)
            .filter(TSBase.datetime >= dateA)
            .filter(TSBase.datetime <= dateB)
            ).all()
    print(qs)
    assert 2 == len(qs)
    print('-'*80)


    # Query-3: select all data (including coordinates) for date A
    dateA = date(2001,1,1)
    qs = (session.query(Location.lat, Location.lon, TSBase.datetime, TSBase.value)
            .join(TSBase, Location.timeseries)
            .filter(func.date(TSBase.datetime) == dateA)
            ).all()
    print(qs)
    # @note: qs is list of tuples; easy export to CSV
    assert 1 == len(qs)
    print('-'*80)


if __name__ == '__main__':
    _test_model()

Alternative-3: a-la BigData

如果您在使用数据库时遇到性能问题，我可能会尝试：

• 仍然像现在一样将数据保存在单独的表/数据库/模式中
• 使用您的数据库引擎提供的“本机”解决方案批量导入数据
• 使用MapReduce-like分析。
  • 在这里，我将继续使用 python 和 sqlalchemy，并实现自己的分布式查询和聚合（或找到现有的东西）。显然，这仅在您不需要直接在数据库上生成这些结果时才有效。

edit-1: Alternative-4: TimeSeries databases

我没有大规模使用它们的经验，但绝对值得考虑。

---

如果你以后能分享你的发现和整个决策过程，那就太好了。

Answer 2

我会避免你上面提到的数据库设计。我对您正在使用的数据知之甚少，但听起来您应该有两个表。一个表用于位置，一个子表用于 location_data。位置表将存储您上面提到的数据，例如坐标和海拔。 location_data 表将存储 location 表中的 location_id 以及您要跟踪的时间序列数据。

这将消除每次添加另一个位置时更改数据库结构和代码更改，并允许您正在查看的查询类型。

Answer 3

两部分：

只使用两张表

没有必要拥有数十或数百个相同的表。只需为location 和location_data 准备一个表，其中每个条目都将锁定到位置。还为 location_id 在 location_data 表上创建索引，以便您进行高效搜索。

不要使用 sqlalchemy 来创建它

我喜欢 sqlalchemy。我每天都使用它。它非常适合管理数据库和添加一些行，但您不想将它用于具有数百万行的初始设置。您想生成一个与 postgres 的“COPY”语句兼容的文件 [http://www.postgresql.org/docs/9.2/static/sql-copy.html] COPY 将让您快速获取大量数据；它是在转储/恢复操作期间使用的。

sqlalchemy 非常适合查询并添加行。如果你有批量操作，你应该使用 COPY。