[草案]非主流实践：从MySQL到PostgreSQL

站长

2024年03月27日 16:27 · 阅读数 6

[草案]非主流实践：从MySQL到PostgreSQL

差Django5.X相关基础以及一个完整的demo后续补上。今天遭不住了，先溜了

叫草案，但是基本已经搞过了效果还行

背景与挑战

在当今的商业环境中，数据管理和处理的效率直接关系到企业的竞争力。我们的核心业务系统依赖于实时数据库处理，这在过去极大地支持了业务的快速发展。然而，随着数据量的激增和业务复杂度的提高，我们发现原有的MySQL数据库在处理大规模数据时，不仅成本上升，而且性能开始受限，尤其是当实时处理与分析任务同时竞争有限的资源时。

为了解决这一挑战，我们进行了详尽的市场和技术研究，最终决定将部分计算任务从实时处理迁移到离线处理。这一战略转变的目的是双重的：一方面，通过离线处理减轻对主数据库的压力，优化系统性能；另一方面，通过迁移至PostgreSQL，我们预计能够获得更高的成本效益和扩展性。

PostgreSQL的选择

PostgreSQL作为一个开源的对象-关系数据库系统，以其高度的可扩展性和符合标准的特性而闻名。它支持复杂的查询、事务完整性和多版本并发控制，这些特性使其成为高并发处理和大数据集合管理的理想选择。此外，PostgreSQL在处理大型数据仓库和分析型任务时表现出色，正符合我们将计算任务转向离线处理的需求。

离线计算的好处

离线计算为我们提供了更大的灵活性和成本效益。它允许我们在非高峰时段处理数据，减少了对实时系统的干扰，同时也降低了运维成本。通过将数据处理和分析任务迁移到离线环境，我们能够实现更深入的数据挖掘和分析，而不会影响到核心业务系统的性能。

迁移计划的重要性

在这次迁移计划中，我们不仅仅关注短期内的成本节约和性能提升，更重要的是为未来的全面迁移和技术升级打下基础。我们的目标是建立一个更加灵活、可扩展和成本效益高的数据处理平台，以支持业务的持续增长和发展。

MySQL BinLog

Binary Log（简称BinLog）是MySQL用于记录数据库中所有更改的日志文件（不包括SELECT和SHOW这样的只读操作）。这些更改包括表的结构变化（如CREATE TABLE、ALTER TABLE等）和数据的增删改操作（如INSERT、UPDATE、DELETE等）。BinLog是MySQL复制的基础，允许数据从一个MySQL数据库服务器复制到另一个服务器，从而实现数据的同步更新。此外，BinLog也是进行数据恢复的重要工具。

要获取尽可能多的数据变更信息，必须正确配置MySQL的binlog。以下是一些关键的配置项：

binlog_format: 设置为ROW，记录每次变更的行数据。
binlog_row_image: 设置为FULL，确保记录所有列的变更前后值。
binlog_rows_query_log_events: 开启，可在binlog中包含每个事件前的原始SQL语句。
binlog_row_metadata (仅在MySQL 8.x中): 设置为FULL，以获取包括字段名称在内的最详细的元数据。

[mysqld]
binlog_format=ROW
binlog_row_image=FULL
binlog_rows_query_log_events=ON
# 仅适用于MySQL 8.x
binlog_row_metadata=FULL

任务计划

第一阶段：数据同步

我们的第一个任务是创建一个与现有MySQL数据库结构相匹配的PostgreSQL数据库。

考虑到资源的限制，我们选择了一个非传统的方法：使用Django ORM来生成表结构，修改pg和mysql的兼容性配置后直接上Django ORM。通过DjangoORM+数据库路由遍历原始mysql表模型，批量查询后写入目标pg。当然，要使用事务保证迁移数据的一致性

尽管这种方法可能看起来有些反常规，但它为我们提供了快速迁移的可能性，尤其是在没有专业DBA或者开发资源的情况下。

不想碰一点老项目代码，这个实验追求速度所以选择这种方式。当然，Django ORM从来没让我失望过^_^

再次提醒，这一步换成你最熟悉的生态去做！！！

一旦表结构就绪，我们开始了数据同步过程。我们利用Django ORM的数据库路由功能，通过分页查询MySQL数据库，并将结果写入PostgreSQL中。这个过程是在一个事务中完成的，以确保数据一致性。考虑到可能的性能问题，我们采取了批处理和并行处理的策略，并实施了必要的错误处理机制。

生成PostgreSQL表结构

首先，您需要确保Django的模型与MySQL中的表结构相匹配。假设您已经有了相应的Django模型，可以使用以下命令来生成PostgreSQL数据库的表结构：

# 伪代码
# 伪代码
# 伪代码

# 在Django项目的settings.py中配置PostgreSQL数据库
DATABASES = {
    'default': {
        # 默认数据库配置（可以是MySQL）
    },
    'postgres': {
        'ENGINE': 'django.db.backends.postgresql',
        'NAME': 'your_postgres_db_name',
        'USER': 'your_postgres_user',
        'PASSWORD': 'your_postgres_password',
        'HOST': 'your_postgres_host',
        'PORT': 'your_postgres_port',
    }
}

# 运行migrate命令时，指定使用PostgreSQL数据库
python manage.py migrate --database=postgres

配置数据库路由

在Django中，您可以通过定义数据库路由来控制模型的数据库操作应该发生在哪个数据库。以下是一个简单的数据库路由器示例：

# 伪代码
# 伪代码
# 伪代码
# 在Django项目中定义一个数据库路由器
class PrimaryReplicaRouter:
    def db_for_read(self, model, **hints):
        """
        Reads go to a randomly-chosen replica.
        """
        return 'default'

    def db_for_write(self, model, **hints):
        """
        Writes always go to primary.
        """
        return 'postgres'

    def allow_relation(self, obj1, obj2, **hints):
        """
        Relations between objects are allowed if both objects are in the primary/replica pool.
        """
        db_list = ('default', 'postgres')
        if obj1._state.db in db_list and obj2._state.db in db_list:
            return True
        return None

    def allow_migrate(self, db, app_label, model_name=None, **hints):
        """
        All non-auth models end up in this pool.
        """
        return True

然后在settings.py中配置您的路由器：

DATABASE_ROUTERS = ['path.to.PrimaryReplicaRouter']

使用数据库路由同步数据快照

以下是一个示例函数，展示了如何使用Django ORM和数据库路由来同步数据：

# 伪代码
# 伪代码
# 伪代码
from django.core.paginator import Paginator
from django.db import transaction
from yourapp.models import YourModel

def sync_table(batch_size=1000):
    # 从默认数据库（MySQL）读取数据
    queryset = YourModel.objects.using('default').all()
    paginator = Paginator(queryset, batch_size)

    for page_num in range(1, paginator.num_pages + 1):
        page = paginator.page(page_num)
        # 在事务中将数据写入PostgreSQL
        with transaction.atomic(using='postgres'):
            for obj in page.object_list:
                obj.save(using='postgres')

事务控制

上面的sync_table函数中已经包含了事务控制的示例。通过使用transaction.atomic()上下文管理器，可以确保在with块中的操作要么全部成功，要么全部失败，从而维护数据一致性。

Django官方文档参考

以下是一些有用的链接，指向Django官方文档中的相关部分：

请注意，上述代码和示例需要根据您的具体情况进行调整。在实际操作中，您可能还需要考虑数据类型转换、索引和约束的创建等问题。如果您的数据模型非常复杂，或者您需要同步大量数据，可能还需要更细致的错误处理和性能优化策略。

第二阶段：持续监听与数据捕获

为了保证数据同步期间的一致性，我们并行启动了第二个任务：监听并解析MySQL的binlog。

这一任务是通过Python的pymysqlreplication库实现的，我们将捕获的binlog中的变更实时写入Kafka队列中。选择Kafka是出于其高吞吐量和可靠性考虑，这对于保证我们的数据传输不丢失至关重要。

当然，还有一点至关重要：一定要注意kafka消息顺序，如果你没有别的办法，同一个表必须保证单生产者、单消费者、单分区

监听 MySQL 的 binlog 并将变更实时写入 Kafka 是一种常见的做法，用于确保在数据迁移过程中能够捕获并应用实时数据变更。这样可以确保在迁移期间，源数据库的任何更新都不会丢失，并且可以在迁移完成后迅速同步到目标数据库。

使用 pymysqlreplication 库来监听 MySQL 的 binlog 是一个很好的选择，因为它能够读取 binlog 并解析出数据变更事件。然后，这些事件可以被推送到 Kafka 队列中，Kafka 由于其高吞吐量和分布式特性，非常适合作为数据传输的中间件。

以下是一个简化的示例，展示如何使用 pymysqlreplication 来监听 MySQL 的 binlog 并将数据变更推送到 Kafka：

# 导入必要的库
from pymysqlreplication import BinLogStreamReader
from pymysqlreplication.row_event import UpdateRowsEvent
from confluent_kafka import Producer
import json

# Kafka 生产者配置，指定Kafka集群的地址
kafka_config = {
    'bootstrap.servers': 'kafka-server1:9092,kafka-server2:9092'
}

# 创建 Kafka 生产者实例
producer = Producer(kafka_config)

# MySQL 连接配置，包括主机地址、端口、用户名和密码
mysql_settings = {
    'host': 'localhost',
    'port': 3306,
    'user': 'root',
    'passwd': 'password'
}

# 创建一个BinLogStreamReader实例，用于连接到MySQL服务器并读取二进制日志
# server_id 是一个唯一的数字，用于标识binlog客户端
# only_events 指定我们只对UpdateRowsEvent（行更新事件）感兴趣
# blocking 设置为True，表示如果当前没有事件，流将会等待直到有新的事件产生
stream = BinLogStreamReader(
    connection_settings=mysql_settings,
    server_id=100,
    only_events=[UpdateRowsEvent],
    blocking=True
)

# 循环读取二进制日志流中的事件
for binlogevent in stream:
    # 对于每个更新行事件，获取事件中所有行的变更
    for row in binlogevent.rows:
        # 提取并存储变更前的行值
        before_values = row['before_values']
        # 提取并存储变更后的行值
        after_values = row['after_values']
        
        # 创建一个字典来存储变更数据
        change_data = {
            'before': before_values,
            'after': after_values
        }
        
        # 将变更数据序列化为 JSON 字符串
        change_data_json = json.dumps(change_data)
        
        # 发送到 Kafka 主题，这里的 'mysql-updates-topic' 是 Kafka 主题名称
        producer.produce('mysql-updates-topic', change_data_json)
        # 确保所有的消息都被发送出去
        producer.flush()

# 当不再需要读取二进制日志时，关闭流
stream.close()

这段代码中，我们首先配置了 Kafka 生产者，然后创建了一个 BinLogStreamReader 实例来监听 MySQL 的 binlog。在捕获到更新事件时，我们将变更前后的数据转换成一个 JSON 字符串，并发送到 Kafka 主题 mysql-updates-topic。

请记得根据你的实际环境调整 MySQL 连接设置和 Kafka 配置。在生产环境中，你可能还需要处理各种异常情况，比如网络问题、Kafka 推送失败等。此外，如果你的 Kafka 生产者配置需要更多的选项（比如安全认证），你也需要在 kafka_config 中相应地添加配置项。

在实际部署时，你可能还需要考虑如何管理 binlog 的读取位置（checkpointing），以便在服务重启后能够从上次停止的位置继续读取。这通常涉及到将读取的最后一个 binlog 位置存储在某个持久化存储中。

Kafka可以保证在特定条件下消息的顺序：

分区内顺序：Kafka保证单个分区内的消息是有序的。也就是说，如果消息按顺序发送到同一个分区，消费者将按发送的顺序接收它们。

同一个生产者：如果你的生产者实例按顺序发送消息到同一个分区，Kafka将会保持这个顺序。

无重试或幂等生产者：如果生产者配置为不进行重试，或者启用了Kafka的幂等特性，即使在网络故障或其他问题导致重试的情况下，也可以保证消息的顺序性。

然而，以下情况可能会破坏顺序性：

多分区：如果消息被发送到多个分区，Kafka不能保证跨分区的顺序。每个分区内部的顺序是保证的，但是从不同分区读取消息时，顺序可能会混乱。

并发生产者：如果有多个生产者实例发送消息到同一个分区，而这些生产者之间没有协调，那么消息的顺序不能保证。

重试导致的重复：在网络问题或其他故障导致生产者重试发送消息时，如果没有幂等或事务性的生产者，可能会导致消息重复或顺序错乱。

消费者并发处理：即使消息在Kafka内部是有序的，消费者应用程序如果并行处理消息，也可能会导致处理顺序与接收顺序不一致。

如果你需要保证跨多个分区的全局顺序，你可能需要设计你的应用程序以发送所有需要顺序保证的消息到同一个分区，并且可能需要在消费端做额外的处理来保证顺序。

第三阶段：实时数据更新

最后，我们的第三个任务是从Kafka读取binlog记录，并将这些更改应用到PostgreSQL数据库中。这个任务与第一个任务共享同一个项目，以便于管理和维护。我们特别注意到所有的更新都应该是幂等的，即多次应用同一个操作结果是不变的。这样，即使在某些极端情况下发生了数据重复或者消息重复消费，我们的系统也能保持稳定。

下面是一个用于实现从消息队列（MQ）中读取binlog日志，并将这些更改实时同步到PostgreSQL数据库的过程。这个过程将与第一个任务整合在同一个项目中，以便提高维护效率和复用性。在实现时，需要特别注意数据的幂等性，以确保在各种极端情况下，如消息重复消费或并发写入，数据的一致性不会被破坏。

设计概要

集成消费者逻辑：在第一个任务的项目中集成一个消息队列消费者，它负责订阅并消费包含binlog日志的消息。
消息解析：消费者接收到消息后，解析这些消息以提取数据库变更信息。这些信息应该包含足够的细节以反映MySQL中的变更，如记录的新状态和唯一标识符。
数据同步：将解析后的变更应用到PostgreSQL数据库。这可能涉及到插入新记录或更新现有记录。需要设计逻辑以确保操作的幂等性，例如使用主键或唯一约束来识别记录。
幂等性保证：在应用变更时，确保重复的操作不会导致数据状态的变化。对于已经存在的记录，可以利用PostgreSQL的ON CONFLICT语句来避免重复插入。
错误处理和日志：实现错误处理机制，以便在遇到问题时重试或记录错误。同时，保持详细的日志记录，以便问题追踪和性能监控。
性能优化：监控数据同步过程的性能，确保系统能够高效地处理消息队列中的数据流。

实现逻辑

消费者服务：在Django项目中实现一个消费者服务，可能是一个命令行管理命令，该命令在后台运行，监听MQ中的binlog更新。
消息处理：每当消费者收到消息，它会解析消息内容，将变更映射到相应的Django模型，并执行数据库操作。
幂等操作：对于每个数据库操作，使用Django ORM的get_or_create和update_or_create方法，或者直接使用SQL语句中的UPSERT操作。
同步逻辑：持续监控MQ中的消息，并同步到PostgreSQL。一旦第一个任务完成了当前快照的同步，它将停止，而第二和第三个任务则继续执行，保持MySQL和PostgreSQL的数据一致性。
异常管理：当出现重复消费或数据冲突时，逻辑应该能够识别并处理这些情况。对于不符合预期的数据变更，应该有一套机制来记录和警告。

关键策略与最佳实践

在这个过程中，我们遵循了几个关键策略和最佳实践：

分页与批处理：以减少单次查询对系统的影响，并提高数据同步的效率。
事务一致性：确保每一步操作都在事务的保护下进行，从而维护数据的一致性。
错误处理：实施适当的错误捕获与重试逻辑，以应对同步过程中可能出现的问题。
性能监控：在迁移过程中监控系统性能，及时调整策略以应对瓶颈。
安全性：使用加密通道和数据脱敏技术来保护数据传输过程中的隐私和安全。
详尽文档：记录每一步的操作和决策，为未来的迁移工作和可能的问题解决提供参考。

参考资料

Django ORM文档
pymysqlreplication库文档
Kafka官方文档
PostgreSQL官方文档
MySQL官方文档