在微服务和分布式架构盛行的今天，很多团队面临一个现实问题：MySQL跑了多年，数据量越来越大，查询开始变慢，但直接迁移到新数据库风险太大。ScyllaDB作为Cassandra的"加速版"，凭借共享无（Shared-Nothing）架构和多核优化，在海量数据写入和读取场景下表现亮眼。但把ScyllaDB直接替换MySQL并不现实——事务支持、生态兼容性、运维习惯，这些都不是简单切换就能解决的。更务实的做法是：让ScyllaDB和MySQL各司其职，通过高效的数据同步和查询优化，把整个系统的性能拉到上限。
这篇文章就讲清楚怎么把ScyllaDB和MySQL对接起来，怎么让数据同步不丢不错，怎么让查询跑得又快又稳。
一、为什么要把ScyllaDB和MySQL放在一块用
1.1 常见的使用场景
先说说什么情况下你会需要这种组合。
第一种场景：MySQL负责"写"，ScyllaDB负责"读"。比如电商系统，订单写入走MySQL保证事务一致性，但订单查询、搜索、统计这些高并发读操作走ScyllaDB。因为ScyllaDB的水平扩展能力更强，加机器就能提升读性能，不用动MySQL的表结构。
第二种场景：实时分析与历史数据归档。MySQL存最近的热数据，ScyllaDB存全量历史数据，通过CDC实时同步增量，查询历史数据时直接查ScyllaDB。
第三种场景：混合负载隔离。业务库和报表库分开，业务用MySQL，报表查询走ScyllaDB，避免报表查询拖慢业务写入。
1.2 数据同步的整体思路
不管用哪种方案，核心思路都是一样的：MySQL作为数据源头，ScyllaDB作为读副本或扩展存储。同步的方式主要有两种——实时同步和批量同步。实时同步靠CDC工具监听MySQL的binlog，把变更实时推送到ScyllaDB；批量同步则适合初次导入或数据修复，定期把MySQL的数据全量灌入ScyllaDB。
二、数据同步方案详解
2.1 基于CDC的实时同步
技术栈：Python 3.10 + Debezium Kafka Connect + Scylla Driver for Python
CDC的全称是Change Data Capture，中文叫"变更数据捕获"。你可以把它理解为一个"监听器"——MySQL每次插入、更新、删除操作，都会在binlog里留一条记录，CDC工具就盯着这个binlog，把每条变更记录实时推送给目标数据库。Debezium是目前最流行的CDC框架之一，支持多种数据源和目的地。
下面是一个完整的CDC同步示例，使用Debezium监听MySQL的binlog，然后通过Python脚本将数据写入ScyllaDB：
# ============================================================
# 技术栈：Python 3.10 + Debezium Kafka Connect + Scylla Driver for Python
# 功能：监听MySQL binlog，将变更数据实时同步到ScyllaDB
# ============================================================
from scylla_driver import ScyllaDriver
import json
import logging
from datetime import datetime
# 配置日志输出，方便调试
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class MySQLToScyllaSync:
"""
MySQL到ScyllaDB的实时同步处理器
通过消费Debezium产出的Kafka消息，写入ScyllaDB
"""
def __init__(self, scylla_host: str, scylla_port: int = 9042):
"""
初始化连接池
:param scylla_host: ScyllaDB节点地址
:param scylla_port: CQL端口号
"""
# 创建连接池，核心连接数3，最大10，空闲超时60秒
self.cluster = ScyllaDriver(
contact_points=[scylla_host],
port=scylla_port,
cores=3,
max_workers=10,
idle_timeout=60
)
self.session = self.cluster.connect('orders_db')
logger.info(f"已连接到ScyllaDB: {scylla_host}:{scylla_port}")
def create_tables(self):
"""初始化ScyllaDB表结构"""
# 订单主表，以order_id为分区键
self.session.execute('''
CREATE TABLE IF NOT EXISTS orders_db.orders (
order_id TEXT PRIMARY KEY,
user_id TEXT,
amount DOUBLE,
status TEXT,
created_at TIMESTAMP,
updated_at TIMESTAMP
) WITH default_time_to_live = 0;
''')
# 按用户查询的辅助表，用户ID作为分区键
self.session.execute('''
CREATE TABLE IF NOT EXISTS orders_db.orders_by_user (
user_id TEXT,
order_id TEXT,
amount DOUBLE,
status TEXT,
created_at TIMESTAMP,
PRIMARY KEY (user_id, order_id)
);
''')
logger.info("ScyllaDB表结构已就绪")
def handle_insert(self, msg: dict):
"""
处理INSERT事件
:param msg: Debezium消息体
"""
after = msg['after']
# 写入主表
self.session.execute('''
INSERT INTO orders_db.orders (order_id, user_id, amount, status, created_at, updated_at)
VALUES (%s, %s, %s, %s, %s, %s)
''', (
after['order_id'],
after['user_id'],
after['amount'],
after['status'],
after['created_at'],
datetime.now()
))
# 写入用户维度的辅助表
self.session.execute('''
INSERT INTO orders_db.orders_by_user (user_id, order_id, amount, status, created_at)
VALUES (%s, %s, %s, %s, %s)
''', (
after['user_id'],
after['order_id'],
after['amount'],
after['status'],
after['created_at']
))
logger.info(f"INSERT处理完成: order_id={after['order_id']}")
def handle_update(self, msg: dict):
"""
处理UPDATE事件
:param msg: Debezium消息体
"""
after = msg['after']
before = msg['before']
self.session.execute('''
UPDATE orders_db.orders SET status = %s, amount = %s, updated_at = %s
WHERE order_id = %s
''', (
after['status'],
after['amount'],
datetime.now(),
after['order_id']
))
logger.info(f"UPDATE处理完成: order_id={after['order_id']}, "
f"状态从 {before['status']} 变为 {after['status']}")
def handle_delete(self, msg: dict):
"""
处理DELETE事件
:param msg: Debezium消息体
"""
before = msg['before']
self.session.execute('''
DELETE FROM orders_db.orders WHERE order_id = %s
''', (before['order_id'],))
self.session.execute('''
DELETE FROM orders_db.orders_by_user WHERE user_id = %s AND order_id = %s
''', (before['user_id'], before['order_id']))
logger.info(f"DELETE处理完成: order_id={before['order_id']}")
def process_message(self, kafka_message):
"""
处理单条Kafka消息
:param kafka_message: Kafka消息对象
"""
msg = json.loads(kafka_message.value)
op = msg['op'] # 操作类型: c=创建, u=更新, d=删除, r=读取
if op == 'c':
self.handle_insert(msg)
elif op == 'u':
self.handle_update(msg)
elif op == 'd':
self.handle_delete(msg)
else:
logger.warning(f"未知操作类型: {op}")
def close(self):
"""关闭连接池"""
self.session.shutdown()
self.cluster.shutdown()
logger.info("ScyllaDB连接已关闭")
# 模拟一条Debezium的INSERT消息（实际从Kafka消费）
sample_message = {
"before": None,
"after": {
"order_id": "ORD-2024-001",
"user_id": "USER-10086",
"amount": 299.99,
"status": "pending",
"created_at": "2024-01-15T10:30:00Z"
},
"op": "c"
}
# 运行同步
if __name__ == '__main__':
sync = MySQLToScyllaSync('192.168.1.100')
sync.create_tables()
sync.process_message(sample_message)
sync.close()
上面这个示例展示了完整的同步流程。Debezium配置部分通常通过Kafka Connect来做，下面给出Kafka Connect的JSON配置：
{
"name": "mysql-orders-cdc",
"config": {
"connector.class": "io.debezium.connector.mysql.MySqlConnector",
"database.hostname": "192.168.1.50",
"database.port": "3306",
"database.user": "cdc_user",
"database.password": "cdc_pass_2024",
"database.server.id": "5400",
"database.server.name": "orders_prod",
"database.include.list": "orders_db",
"table.include.list": "orders_db.orders",
"topic.prefix": "orders_prod",
"schema.history.internal.kafka.topic": "schema.changes.orders_prod",
"schema.history.internal.kafka.bootstrap.servers": "192.168.1.200:9092",
"decimal.handling.mode": "double",
"datetime.precision.mode": "nano"
}
}
2.2 批量导入与全量同步
初次对接ScyllaDB时，通常需要先做一次全量数据导入。如果MySQL里有几千万甚至上亿条数据，一条条INSERT肯定不行，得用批量工具。
技术栈：Python 3.10 + mysql-connector-python + Scylla Driver for Python
# ============================================================
# 技术栈：Python 3.10 + mysql-connector-python + Scylla Driver for Python
# 功能：从MySQL批量读取数据，写入ScyllaDB（全量同步）
# ============================================================
import mysql.connector
from scylla_driver import ScyllaDriver
import logging
from datetime import datetime
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class BatchImporter:
"""批量数据导入工具，分批次从MySQL迁移到ScyllaDB"""
BATCH_SIZE = 5000 # 每批处理5000条数据
def __init__(self):
# MySQL连接配置
self.mysql_conn = mysql.connector.connect(
host='192.168.1.50',
port=3306,
user='migration_user',
password='mig_pass_2024',
database='orders_db',
charset='utf8mb4',
collation='utf8mb4_unicode_ci'
)
self.mysql_cursor = self.mysql_conn.cursor(dictionary=True)
# ScyllaDB连接池
self.cluster = ScyllaDriver(contact_points=['192.168.1.100'])
self.session = self.cluster.connect('orders_db')
def import_orders(self, last_id: int = 0):
"""
分批导入订单数据
:param last_id: 上次导入结束时的最大ID，用于断点续传
"""
total_imported = 0
while True:
# 从MySQL分批查询，每次取BATCH_SIZE条
self.mysql_cursor.execute('''
SELECT id, order_id, user_id, amount, status, created_at
FROM orders
WHERE id > %s
ORDER BY id ASC
LIMIT %s
''', (last_id, self.BATCH_SIZE))
rows = self.mysql_cursor.fetchall()
if not rows:
logger.info("所有数据已同步完成")
break
# 构建批量插入语句
values = []
for row in rows:
values.append((
row['order_id'],
row['user_id'],
row['amount'],
row['status'],
row['created_at']
))
last_id = max(last_id, row['id'])
# 使用批量执行提升性能
self.session.execute_batch([
self.cluster.query('''
INSERT INTO orders_db.orders (order_id, user_id, amount, status, created_at)
VALUES (%s, %s, %s, %s, %s)
''', value)
for value in values
])
total_imported += len(values)
logger.info(f"已同步 {total_imported} 条，当前批次结束ID: {last_id}")
return total_imported
def verify_sync(self, sample_size: int = 100):
"""随机抽样验证数据一致性"""
self.mysql_cursor.execute('''
SELECT order_id, amount, status
FROM orders
ORDER BY RAND()
LIMIT %s
''', (sample_size,))
mismatches = 0
for row in self.mysql_cursor.fetchall():
result = self.session.execute('''
SELECT order_id, amount, status
FROM orders_db.orders
WHERE order_id = %s
''', (row['order_id'],))
row_scylla = result.one()
if not row_scylla or abs(row['amount'] - row_scylla['amount']) > 0.01:
mismatches += 1
logger.warning(f"数据不一致: {row['order_id']}")
logger.info(f"抽样验证完成，{sample_size}条中不一致: {mismatches}条")
return mismatches == 0
def cleanup(self):
"""清理资源"""
self.mysql_cursor.close()
self.mysql_conn.close()
self.session.shutdown()
self.cluster.shutdown()
if __name__ == '__main__':
importer = BatchImporter()
count = importer.import_orders()
importer.verify_sync()
importer.cleanup()
logger.info(f"全量同步完成，共导入 {count} 条数据")
2.3 延迟优化与一致性保障
同步延迟是这类架构中最让人头疼的问题。MySQL那边刚提交的事务，ScyllaDB可能还没同步过来。如果用户紧接着查ScyllaDB，可能读到旧数据。
优化延迟的关键点有三个：
第一，缩短CDC端到端延迟。Debezium本身的延迟通常在几十毫秒级别，但Kafka到消费者这段可能增加几百毫秒甚至更多。减少Kafka分区数、使用poller线程池可以缓解。
第二，利用ScyllaDB的轻量级事务（LWT）处理并发冲突。ScyllaDB不支持MySQL那样的ACID事务，但它的LWT（基于Paxos协议的Compare-And-Swap操作）可以处理简单的条件写入。
# ============================================================
# 技术栈：Python 3.10 + Scylla Driver for Python
# 功能：使用轻量级事务（LWT）处理并发写入冲突
# ============================================================
from scylla_driver import ScyllaDriver
from datetime import datetime
import logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
cluster = ScyllaDriver(contact_points=['192.168.1.100'])
session = cluster.connect('orders_db')
def update_with_lwt(order_id: str, new_status: str, expected_old_status: str):
"""
使用LWT（轻量级事务）更新订单状态
只有当订单当前状态等于expected_old_status时，才允许更新
相当于MySQL中的: UPDATE orders SET status = new_status WHERE status = expected_old_status
:param order_id: 订单ID
:param new_status: 目标状态
:param expected_old_status: 期望的旧状态（用于条件判断）
:return: True表示更新成功，False表示状态冲突
"""
result = session.execute('''
UPDATE orders_db.orders
SET status = %s, updated_at = %s
WHERE order_id = %s
IF status = %s
''', (new_status, datetime.now(), order_id, expected_old_status))
# ScyllaDB的LWT返回_applys列表示是否成功应用
if result and result[0] == True:
logger.info(f"LWT更新成功: order_id={order_id}, 状态变为 {new_status}")
return True
else:
# 获取当前值，帮助排查冲突原因
current = session.execute(
'SELECT status FROM orders_db.orders WHERE order_id = %s', (order_id,)
).one()
logger.warning(
f"LWT更新失败: order_id={order_id}, "
f"期望状态={expected_old_status}, 实际状态={current['status'] if current else '不存在'}"
)
return False
# 示例：将订单从"待支付"更新为"已支付"
success = update_with_lwt("ORD-2024-001", "paid", "pending")
cluster.shutdown()
第三，引入最终一致性校验机制。定时任务每隔一段时间对比MySQL和ScyllaDB的数据量，发现差异就触发补同步。这就像定期"对账"一样。
三、查询效率优化策略
3.1 查询模式分析与索引设计
ScyllaDB和MySQL在查询模型上有一个根本区别：ScyllaDB没有通用的二级索引（虽然它有，但性能不如表设计来得好）。ScyllaDB强调的是"把你要查的东西提前按查询模式组织好"。
简单说就是：你想怎么查数据，就怎么建表。
举个例子，如果业务需要按用户ID查订单列表，那就要建一个以user_id为分区键的表。如果还要按订单状态筛选，那status可以作为排序键的一部分。
# ============================================================
# 技术栈：Python 3.10 + Scylla Driver for Python
# 功能：展示ScyllaDB的表设计策略与查询模式
# ============================================================
from scylla_driver import ScyllaDriver
import logging
from datetime import datetime
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
cluster = ScyllaDriver(contact_points=['192.168.1.100'])
session = cluster.connect('orders_db')
# --- 查询模式1：按用户ID查全部订单 ---
session.execute('''
CREATE TABLE IF NOT EXISTS orders_db.orders_by_user (
user_id TEXT,
order_id TEXT,
amount DOUBLE,
status TEXT,
created_at TIMESTAMP,
PRIMARY KEY (user_id, order_id)
) WITH CLUSTERING ORDER BY (order_id ASC);
''')
# 查询该用户最近10个订单
result = session.execute('''
SELECT order_id, amount, status, created_at
FROM orders_db.orders_by_user
WHERE user_id = %s
ORDER BY order_id DESC
LIMIT 10
''', ('USER-10086',))
for row in result:
logger.info(f"订单: {row['order_id']}, 金额: {row['amount']}, 状态: {row['status']}")
# --- 查询模式2：按用户+时间范围查订单 ---
session.execute('''
CREATE TABLE IF NOT EXISTS orders_db.orders_by_user_time (
user_id TEXT,
created_at TIMESTAMP,
order_id TEXT,
amount DOUBLE,
status TEXT,
PRIMARY KEY (user_id, created_at, order_id)
) WITH CLUSTERING ORDER BY (created_at DESC, order_id DESC);
''')
# 查询某用户2024年1月的全部订单
result = session.execute('''
SELECT order_id, amount, status, created_at
FROM orders_db.orders_by_user_time
WHERE user_id = %s AND created_at >= %s
LIMIT 50
''', ('USER-10086', datetime(2024, 1, 1)))
logger.info(f"2024年1月订单数: {len(result.all())}")
# --- 查询模式3：按订单状态查订单 ---
session.execute('''
CREATE TABLE IF NOT EXISTS orders_db.orders_by_status (
status TEXT,
created_at TIMESTAMP,
order_id TEXT,
user_id TEXT,
amount DOUBLE,
PRIMARY KEY (status, created_at, order_id)
) WITH CLUSTERING ORDER BY (created_at DESC, order_id DESC);
''')
# 查询所有"待支付"状态的订单
result = session.execute('''
SELECT order_id, user_id, amount, created_at
FROM orders_db.orders_by_status
WHERE status = 'pending'
LIMIT 100
''')
for row in result:
logger.info(f"待支付订单: {row['order_id']}, 用户: {row['user_id']}, 金额: {row['amount']}")
cluster.shutdown()
3.2 预聚合与物化视图
ScyllaDB支持物化视图（Materialized View），这是把预计算结果存储起来的概念。和MySQL的视图不同，MySQL视图是虚拟的，每次查询都会重新执行SQL；ScyllaDB的物化视图是真正建一张新表，把数据提前存好。
但要注意，ScyllaDB的物化视图有使用限制——更新频率高的表不建议建，因为每次更新都要同时更新物化视图，会拖慢写入性能。
# ============================================================
# 技术栈：Python 3.10 + Scylla Driver for Python
# 功能：使用ScyllaDB物化视图实现预聚合查询
# ============================================================
from scylla_driver import ScyllaDriver
import logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
cluster = ScyllaDriver(contact_points=['192.168.1.100'])
session = cluster.connect('orders_db')
# 先确保基础表存在
session.execute('''
CREATE TABLE IF NOT EXISTS orders_db.orders (
order_id TEXT PRIMARY KEY,
user_id TEXT,
amount DOUBLE,
status TEXT,
created_at TIMESTAMP,
region TEXT
);
''')
# 创建物化视图：按地区聚合订单金额
# 注意：物化视图的WHERE子句里必须包含主键（分区键+聚簇键）
session.execute('''
CREATE MATERIALIZED VIEW IF NOT EXISTS orders_db.orders_by_region AS
SELECT order_id, user_id, amount, status, created_at, region
FROM orders_db.orders
WHERE order_id IS NOT NULL AND region IS NOT NULL
PRIMARY KEY (region, order_id)
WITH CLUSTERING ORDER BY (order_id DESC)
AND default_time_to_live = 0;
''')
logger.info("物化视图已创建")
# 查询华东地区的所有订单
result = session.execute('''
SELECT order_id, user_id, amount, status
FROM orders_db.orders_by_region
WHERE region = 'east_china'
LIMIT 20
''')
total_amount = sum(row['amount'] for row in result)
logger.info(f"华东地区订单总数: {len(result.all())}, 总金额: {total_amount:.2f}")
# 物化视图清理：不再需要时
# session.execute('DROP MATERIALIZED VIEW orders_db.orders_by_region');
# logger.info("物化视图已删除")
cluster.shutdown()
3.3 连接池与缓存策略
连接池配置直接影响查询性能。ScyllaDB的Python驱动内置了连接池管理，但需要合理配置参数。
下面是一个完整的查询服务示例，包含连接池配置和本地缓存：
# ============================================================
# 技术栈：Python 3.10 + Scylla Driver for Python
# 功能：配置连接池参数与本地缓存，优化查询性能
# ============================================================
from scylla_driver import ScyllaDriver
from functools import lru_cache
import logging
from datetime import datetime, timedelta
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class QueryService:
"""
ScyllaDB查询服务
包含连接池调优和本地结果缓存
"""
def __init__(self):
# 连接池配置说明：
# cores=4：核心线程数，建议设置为CPU核数的1/4
# max_workers=20：最大工作线程，高并发场景可设大一些
# idle_timeout=120：空闲连接超时时间（秒）
# heartbeat_interval=30：心跳间隔，用于检测死连接
# reconnect_backoff: 重连退避策略
self.cluster = ScyllaDriver(
contact_points=['192.168.1.100', '192.168.1.101', '192.168.1.102'],
port=9042,
cores=4,
max_workers=20,
idle_timeout=120,
heartbeat_interval=30,
consistency_level='LOCAL_QUORUM', # 一致性级别：本地多数
)
self.session = self.cluster.connect('orders_db')
logger.info("查询服务初始化完成，连接池已就绪")
def get_user_orders(self, user_id: str, limit: int = 50):
"""
查询用户订单列表
一致性级别LOCAL_QUORUM：至少半数节点确认写入才返回成功
如果追求更低延迟，可以降为ONE
"""
query = '''
SELECT order_id, amount, status, created_at
FROM orders_db.orders_by_user
WHERE user_id = %s
ORDER BY order_id DESC
LIMIT %s
'''
start_time = datetime.now()
rows = self.session.execute(query, (user_id, limit)).all()
elapsed = (datetime.now() - start_time).total_seconds() * 1000
logger.info(f"查询用户订单: user_id={user_id}, 返回 {len(rows)} 条, 耗时 {elapsed:.1f}ms")
return rows
def get_order_detail(self, order_id: str):
"""
查询单个订单详情
主键查询，性能最优
"""
result = self.session.execute('''
SELECT * FROM orders_db.orders WHERE order_id = %s
''', (order_id,))
row = result.one()
return dict(row) if row else None
def get_hot_orders(self, hours: int = 24):
"""
查询最近N小时的热门订单（带缓存）
使用lru_cache做本地缓存，30秒过期
"""
# 缓存最大1000个结果，每条缓存30秒
cached = self._get_hot_orders_cached(hours)
return cached
@lru_cache(maxsize=1000)
def _get_hot_orders_cached(self, hours: int):
"""内部缓存方法"""
cutoff_time = datetime.now() - timedelta(hours=hours)
rows = self.session.execute('''
SELECT order_id, user_id, amount, status, created_at
FROM orders_db.orders_by_status
WHERE status = 'paid' AND created_at >= %s
ORDER BY created_at DESC
LIMIT 100
''', (cutoff_time,)).all()
return [dict(r) for r in rows]
def close(self):
"""优雅关闭连接池"""
self.session.shutdown()
self.cluster.shutdown()
logger.info("查询服务已关闭")
# 使用示例
if __name__ == '__main__':
service = QueryService()
# 查询用户订单
orders = service.get_user_orders('USER-10086', limit=20)
# 查询订单详情
detail = service.get_order_detail('ORD-2024-001')
# 查询热门订单
hot_orders = service.get_hot_orders(hours=24)
service.close()
四、技术优缺点分析
4.1 优势
这套方案的好处非常明显。首先是性能隔离——MySQL负责事务和写入，ScyllaDB负责高并发读取，互不干扰。其次是扩展性——ScyllaDB加机器就能横向扩展，不用动MySQL的架构。第三是数据安全性——MySQL作为数据源头，ScyllaDB只是副本，即使ScyllaDB出问题，数据也不会丢。
4.2 不足
代价也是存在的。第一，架构复杂度上升，多了一套基础设施要维护。第二，同步延迟不可避免，对强一致性要求高的场景不太合适。第三，ScyllaDB的学习曲线比MySQL陡峭，开发团队需要花时间理解分区键、一致性级别这些概念。第四，存储成本翻倍，同样的数据存了两份。
五、注意事项与最佳实践
5.1 数据类型映射
MySQL和ScyllaDB的数据类型不完全一样，映射的时候要注意。比如MySQL的DATETIME对应ScyllaDB的TIMESTAMP，MySQL的DECIMAL可以映射为DOUBLE（会损失精度）或者TEXT（保持精度但无法直接做数值计算）。
# ============================================================
# 技术栈：Python 3.10 + Scylla Driver for Python + mysql-connector-python
# 功能：MySQL到ScyllaDB的数据类型映射处理
# ============================================================
from scylla_driver import ScyllaDriver
import mysql.connector
from datetime import datetime
import decimal
import logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
cluster = ScyllaDriver(contact_points=['192.168.1.100'])
session = cluster.connect('orders_db')
def convert_datatype(mysql_row: dict) -> tuple:
"""
将MySQL行数据转换为ScyllaDB兼容的类型
处理常见类型差异：
- DECIMAL -> float（接受少量精度损失）
- DATETIME -> datetime对象
- BLOB -> bytes
- ENUM/SET -> TEXT
"""
return (
str(mysql_row['order_id']), # VARCHAR -> TEXT
str(mysql_row['user_id']), # VARCHAR -> TEXT
float(mysql_row['amount']), # DECIMAL(10,2) -> DOUBLE
str(mysql_row['status']), # ENUM -> TEXT
mysql_row['created_at'], # DATETIME -> TIMESTAMP
mysql_row['remark'].encode('utf-8') # BLOB/TEXT -> BLOB
if mysql_row.get('remark') else None
)
# 示例：从MySQL读取并转换后写入ScyllaDB
conn = mysql.connector.connect(
host='192.168.1.50', user='reader', password='pass', database='orders_db'
)
cursor = conn.cursor(dictionary=True)
cursor.execute('SELECT * FROM orders WHERE id = 1 LIMIT 1')
row = cursor.fetchone()
if row:
values = convert_datatype(row)
session.execute('''
INSERT INTO orders_db.orders (order_id, user_id, amount, status, created_at)
VALUES (%s, %s, %s, %s, %s)
''', values[:5])
logger.info(f"数据转换并写入成功: order_id={values[0]}")
cursor.close()
conn.close()
cluster.shutdown()
5.2 错误处理与重试机制
网络抖动、节点宕机这些情况在分布式系统里是常态。写ScyllaDB的代码必须加上重试和降级逻辑。
# ============================================================
# 技术栈：Python 3.10 + Scylla Driver for Python
# 功能：带重试和降级的ScyllaDB写入封装
# ============================================================
from scylla_driver import ScyllaDriver
from cassandra import ConsistencyLevel
import logging
import time
import random
from datetime import datetime
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
cluster = ScyllaDriver(
contact_points=['192.168.1.100', '192.168.1.101', '192.168.1.102'],
consistency_level=ConsistencyLevel.LOCAL_QUORUM
)
session = cluster.connect('orders_db')
MAX_RETRIES = 3 # 最大重试次数
BACKOFF_BASE = 0.5 # 退避基数（秒）
def safe_insert(order_data: dict) -> bool:
"""
安全写入订单数据，自动重试
使用指数退避策略，每次重试等待时间翻倍
如果全部重试失败，降级为低一致性写入
"""
for attempt in range(MAX_RETRIES):
try:
session.execute('''
INSERT INTO orders_db.orders (order_id, user_id, amount, status, created_at)
VALUES (%s, %s, %s, %s, %s)
''', (
order_data['order_id'],
order_data['user_id'],
order_data['amount'],
order_data['status'],
order_data['created_at']
))
logger.info(f"写入成功: order_id={order_data['order_id']}, 尝试次数={attempt+1}")
return True
except Exception as e:
# 计算等待时间：0.5s, 1s, 2s（指数退避 + 随机抖动）
wait_time = (BACKOFF_BASE ** attempt) * (1 + random.random())
logger.warning(f"第 {attempt+1} 次尝试失败: {str(e)}, {wait_time:.2f}秒后重试")
time.sleep(wait_time)
# 所有重试失败，降级为低一致性写入（ONE）
logger.error(f"高一致性写入全部失败，尝试降级写入: order_id={order_data['order_id']}")
try:
statement = '''
INSERT INTO orders_db.orders (order_id, user_id, amount, status, created_at)
VALUES (%s, %s, %s, %s, %s)
'''
statement = cluster.prepare(statement)
# 设置一致性级别为ONE，只等一个节点确认
statement.consistency_level = ConsistencyLevel.ONE
session.execute(statement, (
order_data['order_id'],
order_data['user_id'],
order_data['amount'],
order_data['status'],
order_data['created_at']
))
logger.warning(f"降级写入成功（一致性级别: ONE）: order_id={order_data['order_id']}")
return True
except Exception as final_error:
logger.critical(f"所有写入均失败: {str(final_error)}")
return False
# 测试写入
test_order = {
"order_id": "ORD-2024-TEST-001",
"user_id": "USER-TEST",
"amount": 150.00,
"status": "pending",
"created_at": datetime.now()
}
safe_insert(test_order)
cluster.shutdown()
六、应用场景总结
这套ScyllaDB与MySQL集成方案，最适合以下几类业务：
高并发读场景——比如电商的订单查询、社交平台的动态feed、物联网的设备状态查询，这些场景读多写少，ScyllaDB的水平扩展能力可以完美应对。
海量数据存储场景——日志数据、传感器数据、用户行为数据，这些数据写入频繁且总量巨大，MySQL扛不住，ScyllaDB能轻松应对。
混合负载隔离场景——业务系统和报表系统分开部署，业务用MySQL保证事务，报表用ScyllaDB保证查询速度，两边互不干扰。
但如果是金融交易这类对强一致性要求极高的场景，或者数据量不大、MySQL性能还够用的场景，就不推荐引入ScyllaDB。多一套组件就多一份维护成本，没有足够的收益就没必要。
总的来说，ScyllaDB和MySQL不是替代关系，而是互补关系。MySQL在事务处理和复杂查询方面依然是王者，ScyllaDB在海量数据和水平扩展方面无可替代。把两者用得好，才是架构设计的关键——不是哪个技术先进就用哪个，而是根据业务需求选对组合。
评论
围绕“ScyllaDB与传统数据库集成的优化方案：对接MySQL时的数据同步与查询效率”参与讨论