使用Apache Airflow从Oracle同步数据到InfluxDB仅插入单行的问题排查

核心问题定位 #

你遇到的每个company_id仅保留最后一行数据的问题，大概率是以下两种原因导致：

1. InfluxDB数据覆盖机制：当多条数据的measurement、tag集合完全相同，且时间戳（time）一致时，InfluxDB会自动覆盖旧数据，仅保留最后一条。
2. Python代码逻辑错误：复用了同一个Point对象，循环中不断更新该对象的字段，最终仅写入最后一次循环的结果。

代码修复方案 #

错误代码示例（常见问题场景） #

假设你的原有代码存在类似问题：

from influxdb_client import InfluxDBClient, Point
from airflow.providers.oracle.hooks.oracle import OracleHook

def sync_oracle_to_influx():
    oracle_hook = OracleHook(oracle_conn_id='oracle_conn')
    conn = oracle_hook.get_conn()
    cursor = conn.cursor()
    cursor.execute("SELECT company_id, metric_value, record_time FROM company_metrics")
    rows = cursor.fetchall()

    # 问题点：复用同一个Point对象，循环中覆盖数据
    point = Point("company_metrics")
    with InfluxDBClient(url="http://influxdb:8086", token="token", org="my_org") as client:
        write_api = client.write_api()
        for row in rows:
            company_id, metric_value, record_time = row
            point.tag("company_id", company_id).field("metric_value", metric_value).time(record_time)
            write_api.write(bucket="my_bucket", record=point)

修复后的完整代码 #

from influxdb_client import InfluxDBClient, Point
from airflow.providers.oracle.hooks.oracle import OracleHook
from datetime import timedelta

def sync_oracle_to_influx():
    # 1. 获取Oracle数据
    oracle_hook = OracleHook(oracle_conn_id='oracle_conn')
    conn = oracle_hook.get_conn()
    cursor = conn.cursor()
    cursor.execute("SELECT company_id, metric_value, record_time FROM company_metrics")
    rows = cursor.fetchall()
    print(f"Oracle查询到{len(rows)}条数据")

    # 2. 构建InfluxDB数据点列表
    points = []
    for idx, row in enumerate(rows):
        company_id, metric_value, record_time = row
        
        # 处理同一company_id下时间戳重复的情况：添加微秒偏移避免覆盖
        adjusted_time = record_time + timedelta(microseconds=idx)
        
        # 每次循环新建Point对象，避免数据覆盖
        point = Point("company_metrics") \
            .tag("company_id", company_id) \
            .field("metric_value", metric_value) \
            .time(adjusted_time)
        points.append(point)

    # 3. 批量写入InfluxDB
    with InfluxDBClient(url="http://influxdb:8086", token="token", org="my_org") as client:
        write_api = client.write_api()
        write_api.write(bucket="my_bucket", record=points)
        print(f"成功写入InfluxDB{len(points)}条数据")

    # 4. 关闭数据库连接
    cursor.close()
    conn.close()

关键修复说明 #

• 避免Point对象复用：每次循环创建新的Point实例，确保每条数据都是独立的写入单元。
• 处理重复时间戳：通过添加微秒级偏移量，解决同一company_id下时间戳相同导致的数据覆盖问题。
• 批量写入优化：先收集所有数据点再批量写入，既提升效率，也避免单条写入时的潜在异常。

验证步骤 #

1. 代码执行前，打印查询到的Oracle数据行数，确认数据完整性。
2. 写入完成后，在InfluxDB中执行查询：SELECT * FROM company_metrics WHERE company_id = '目标ID'，检查返回行数是否与Oracle中对应company_id的行数一致。

内容的提问来源于stack exchange，提问作者Enamul Haque