数仓建模：增量更新还是全量更新，该如何选择？

大数据量（TB级以上）：倾向于增量更新
小数据量：可以考虑全量更新

高频更新（每小时或更频繁）：增量更新
低频更新（每天或更少）：全量更新可能更简单

高变化率（>30%数据经常变化）：全量更新可能更简单
低变化率：增量更新更有效

资源受限：增量更新
资源充足：可以考虑全量更新

极高一致性要求：可能需要全量更新
可以容忍短暂不一致：增量更新更灵活

简单模型：两种方法都可以
复杂模型（多表关联、复杂转换）：增量更新可能更具挑战性

需要详细的历史记录：增量更新更适合
只关注当前状态：全量更新足够

某些工具可能更适合特定的更新策略

import pandas as pd
from sqlalchemy import create_engine
from datetime import datetime, timedelta


def incremental_order_update(db_engine, last_update_time):
    # 从源系统获取新增和变更的订单数据
    query = f"""
    SELECT order_id, customer_id, order_status, order_amount, order_time
    FROM source_orders
    WHERE order_time >= '{last_update_time}'
       OR (order_status_update_time >= '{last_update_time}' AND order_status_update_time > order_time)
    """
    new_orders = pd.read_sql(query, db_engine)


    # 更新数据仓库
    with db_engine.begin() as conn:
        # 插入新订单
        new_orders.to_sql('dw_orders', conn, if_exists='append', index=False)


        # 更新已存在的订单状态
        for _, row in new_orders.iterrows():
            conn.execute(f"""
            UPDATE dw_orders
            SET order_status = '{row['order_status']}'
            WHERE order_id = {row['order_id']}
            """)


    print(f"Incremental update completed. {len(new_orders)} orders processed.")


# 示例使用
db_engine = create_engine('postgresql://username:password@localhost:5432/datawarehouse')
last_update_time = datetime.now() - timedelta(days=1)
incremental_order_update(db_engine, last_update_time)

import pandas as pd
from sqlalchemy import create_engine


def full_order_update(db_engine):
    # 从源系统获取所有订单数据
    query = """
    SELECT order_id, customer_id, order_status, order_amount, order_time
    FROM source_orders
    """
    all_orders = pd.read_sql(query, db_engine)


    # 更新数据仓库
    with db_engine.begin() as conn:
        # 清空现有数据
        conn.execute("TRUNCATE TABLE dw_orders")


        # 插入所有订单
        all_orders.to_sql('dw_orders', conn, if_exists='append', index=False)


    print(f"Full update completed. {len(all_orders)} orders processed.")


# 示例使用
db_engine = create_engine('postgresql://username:password@localhost:5432/datawarehouse')
full_order_update(db_engine)

数据量大且持续增长：每天100万新订单，全量更新将变得越来越慢
需要支持实时报表：增量更新可以更频繁地执行，提供近实时的数据
历史趋势分析需求：增量更新便于保留和跟踪订单状态的历史变化

日常使用增量更新保持数据的及时性
定然而，我们也可以考虑结合两种方法：
日常使用增量更新保持数据的及时性
定期（如每周或每月）执行一次全量更新，以确保数据的完整性和一致性

-- 为订单表创建合适的索引
CREATE INDEX idx_order_time ON dw_orders(order_time);
CREATE INDEX idx_order_status ON dw_orders(order_status);
CREATE INDEX idx_customer_id ON dw_orders(customer_id);

-- 创建按日期分区的订单表
CREATE TABLE dw_orders (
    order_id INT,
    customer_id INT,
    order_status VARCHAR(20),
    order_amount DECIMAL(10,2),
    order_time TIMESTAMP
) PARTITION BY RANGE (order_time);


-- 创建每月分区
CREATE TABLE dw_orders_y2023m01 PARTITION OF dw_orders
    FOR VALUES FROM ('2023-01-01') TO ('2023-02-01');


CREATE TABLE dw_orders_y2023m02 PARTITION OF dw_orders
    FOR VALUES FROM ('2023-02-01') TO ('2023-03-01');


-- ... 其他月份的分区

def batch_incremental_update(db_engine, batch_size=1000):
    last_processed_id = 0
    while True:
        # 获取一批数据
        batch = pd.read_sql(f"""
            SELECT * FROM source_orders
            WHERE order_id > {last_processed_id}
            ORDER BY order_id
            LIMIT {batch_size}
        """, db_engine)


        if batch.empty:
            break


        # 处理这批数据
        with db_engine.begin() as conn:
            batch.to_sql('dw_orders', conn, if_exists='append', index=False)


        last_processed_id = batch['order_id'].max()
        print(f"Processed batch up to order_id {last_processed_id}")

from concurrent.futures import ThreadPoolExecutor
import pandas as pd


def update_partition(partition_date, db_engine):
    query = f"""
    SELECT * FROM source_orders
    WHERE order_time >= '{partition_date}' AND order_time < '{partition_date + timedelta(days=1)}'
    """
    partition_data = pd.read_sql(query, db_engine)


    with db_engine.begin() as conn:
        partition_data.to_sql(f'dw_orders_{partition_date.strftime("%Y%m%d")}', 
                              conn, if_exists='replace', index=False)


def parallel_full_update(db_engine, start_date, end_date):
    dates = pd.date_range(start_date, end_date)
    with ThreadPoolExecutor(max_workers=4) as executor:
        executor.map(lambda date: update_partition(date, db_engine), dates)


# 使用示例
start_date = datetime(2023, 1, 1)
end_date = datetime(2023, 12, 31)
parallel_full_update(db_engine, start_date, end_date)

def safe_incremental_update(db_engine, data):
    with db_engine.begin() as conn:
        for _, row in data.iterrows():
            while True:
                try:
                    conn.execute("""
                        UPDATE dw_orders
                        SET order_status = %s
                        WHERE order_id = %s AND update_time < %s
                    """, (row['order_status'], row['order_id'], row['update_time']))
                    break
                except sqlalchemy.exc.OperationalError as e:
                    if 'deadlock detected' in str(e):
                        print(f"Deadlock detected for order {row['order_id']}, retrying...")
                        time.sleep(0.1)  # 短暂休眠后重试
                    else:
                        raise

def incremental_update_with_checkpoint(db_engine, batch_size=1000):
    checkpoint = get_last_checkpoint()  # 从某个存储中获取上次的检查点


    while True:
        batch = get_next_batch(checkpoint, batch_size)  # 获取下一批数据
        if not batch:
            break


        try:
            with db_engine.begin() as conn:
                update_data(conn, batch)  # 更新数据
                update_checkpoint(conn, batch[-1]['id'])  # 更新检查点
        except Exception as e:
            print(f"Error occurred: {e}. Rolling back to last checkpoint.")
            # 错误发生时回滚到上一个检查点


    # 更新完成后进行全量校验
    validate_data_consistency(db_engine)

def hybrid_update_strategy(db_engine):
    current_time = datetime.now()


    # 每天执行增量更新
    if current_time.hour == 1:  # 假设在每天凌晨1点执行
        incremental_update(db_engine)


    # 每周日执行全量更新
    if current_time.weekday() == 6 and current_time.hour == 2:
        full_update(db_engine)


    # 每月最后一天执行数据校验
    last_day_of_month = (current_time.replace(day=1) + timedelta(days=32)).replace(day=1) - timedelta(days=1)
    if current_time.date() == last_day_of_month.date() and current_time.hour == 3:
        validate_data_consistency(db_engine)

极低的延迟：数据几乎可以实时反映在报表和分析中。
资源利用更均匀：避免了传统批处理方式的资源使用峰值。
更好的用户体验：为基于数据的实时决策提供支持

流处理框架：如Apache Kafka、Apache Flink等。
变更数据捕获（CDC）：直接从数据库事务日志中捕获变更。
内存数据网格：如Apache Ignite，提供内存中的数据处理能力。

from kafka import KafkaConsumer
from json import loads


consumer = KafkaConsumer(
    'order_topic',
     bootstrap_servers=['localhost:9092'],
     auto_offset_reset='earliest',
     enable_auto_commit=True,
     group_id='order-processing-group',
     value_deserializer=lambda x: loads(x.decode('utf-8'))
)


def process_order(order):
    # 处理订单数据
    with db_engine.begin() as conn:
        conn.execute("""
            INSERT INTO dw_orders (order_id, customer_id, order_status, order_amount, order_time)
            VALUES (%s, %s, %s, %s, %s)
            ON CONFLICT (order_id) DO UPDATE
            SET order_status = EXCLUDED.order_status,
                order_amount = EXCLUDED.order_amount
        """, (order['order_id'], order['customer_id'], order['order_status'], 
              order['order_amount'], order['order_time']))


for message in consumer:
    order = message.value
    process_order(order)

系统复杂性增加：需要管理和维护实时处理管道。
一致性保证更困难：在分布式系统中确保数据一致性变得更加复杂。
错误处理和恢复：实时系统需要更健壮的错误处理机制。

增量更新适合数据量大、变化频繁、需要近实时更新的场景。它能提供更好的性能和更低的资源消耗，但实现复杂度较高。




全量更新适合数据量较小、变化不频繁、对一致性要求高的场景。它实现简单，确保数据完整性，但对大型数据集可能效率较低。




混合策略结合了两者的优点，可以在日常使用增量更新，定期进行全量更新和数据校验。




实时更新是未来的趋势，适合对数据时效性要求极高的场景，但也带来了更高的系统复杂性。

评估数据特征：包括数据量、更新频率、变化程度等。
分析业务需求：考虑数据时效性、一致性、历史追溯等需求。
权衡系统资源：评估可用的计算资源、存储容量和网络带宽。
考虑技术能力：评估团队实现和维护各种更新策略的能力。
进行性能测试：在实际或模拟环境中测试不同策略的性能。
制定监控和应急方案：无论选择哪种策略，都要有完善的监控和问题处理机制。