4、数据采集与接入：SCADA数据接口解析、CSV/Excel批量导入、数据库连接（MySQL、InfluxDB）、API数据获取

各位同学，咱们今天聊点实在的。

做风电大数据，第一步不是建模，也不是画图。第一步是什么？是把数据搞进来。数据进不来，后面全是空谈。我见过太多项目，算法模型吹得天花乱坠，结果数据接口三天没调通，整个项目直接卡死。所以这一章，咱们就把数据采集与接入这件事，掰开了揉碎了讲清楚。

4.1 SCADA数据接口解析

SCADA系统，说白了就是风机的“黑匣子”。它记录了每一台风机的运行状态——风速、转速、桨距角、发电机温度、有功功率……这些数据，是我们做分析的基础。

SCADA的数据接口，常见的有两种：

  • OPC DA/AE：老牌协议，Windows生态，实时性好。但我个人建议，能不用就别用——它依赖DCOM，配置起来能让你怀疑人生。我曾经在一个项目中，光配OPC的防火墙规则就花了两天。
  • OPC UA：跨平台，安全性高，支持加密。现在新项目基本都推OPC UA。如果你有机会选型，直接上UA，别犹豫。
  • Modbus TCP：简单粗暴，很多老旧风机还在用。数据量不大，但胜在稳定。

核心要点：SCADA数据通常是秒级或毫秒级采集。你拿到手的数据，往往是“脏”的——有空值、有跳变、有时间戳错乱。所以，接口解析的第一步，不是读数据，而是做数据质量检查。

举个例子，我解析过的一个SCADA接口，返回的JSON结构大概长这样：

{
  "turbine_id": "WTG-001",
  "timestamp": "2024-03-15T10:30:00Z",
  "measurements": {
    "wind_speed": 12.5,
    "rotor_speed": 14.2,
    "blade_pitch": 3.1,
    "active_power": 1850.0,
    "generator_temp": 68.3
  }
}

嗯，这里要注意：时间戳一定要统一成UTC或北京时间，别混用。我见过一个项目，SCADA用UTC，数据库用北京时间，结果分析出来的功率曲线全是乱的。

4.2 CSV/Excel批量导入

说实话，CSV和Excel是风电行业最“土”但最常用的数据交换格式。为什么？因为很多风电场运维人员，习惯把数据导出成Excel，然后发给你。你总不能说“我不收Excel”吧？

批量导入，有几个坑你得避开：

  • 编码问题：CSV文件经常是GBK编码，而你的Python默认是UTF-8。不处理的话，中文列名直接乱码。
  • 数据类型推断：Excel里“2024-01-01”看起来是日期，但pandas读进来可能变成字符串。你想想看，如果时间列是字符串，后面做时序分析就全废了。
  • 大文件处理：一个CSV文件几百兆，直接read_csv会爆内存。我建议用chunksize分批读取。

我的习惯：批量导入前，先写一个数据预览函数，只读前100行。确认列名、数据类型、时间格式都对，再全量导入。这一步能省你后面80%的调试时间。

代码示例，批量导入CSV并自动处理编码：

import pandas as pd
import glob

def batch_import_csv(folder_path):
    all_files = glob.glob(folder_path + "/*.csv")
    df_list = []
    for file in all_files:
        # 先尝试UTF-8，不行就GBK
        try:
            df = pd.read_csv(file, encoding='utf-8')
        except UnicodeDecodeError:
            df = pd.read_csv(file, encoding='gbk')
        df_list.append(df)
    return pd.concat(df_list, ignore_index=True)

4.3 数据库连接：MySQL与InfluxDB

数据量大了以后，文件存不住。这时候就得上数据库。

MySQL：关系型数据库，适合存结构化数据。比如风机台账、故障记录、月度报表。我建议把SCADA的元数据（风机编号、型号、投运日期）放MySQL，查询快，关联方便。

InfluxDB：时序数据库，专门为时间序列数据设计的。风电的秒级数据，用InfluxDB存，查询性能比MySQL高一个数量级。而且它自带数据保留策略，可以自动清理过期数据。

避坑指南：我曾经在一个项目中，把SCADA秒级数据全塞进MySQL。结果一个月后，单表数据量超过5亿行，查询一条功率曲线要等30秒。后来换成InfluxDB，同样的查询只要0.3秒。所以，时序数据，请用时序数据库。

连接MySQL的示例：

import pymysql

conn = pymysql.connect(
    host='localhost',
    user='wind_user',
    password='your_password',
    database='wind_farm'
)
cursor = conn.cursor()
cursor.execute("SELECT * FROM turbine_info WHERE farm_id = 'FARM-001'")
results = cursor.fetchall()

连接InfluxDB的示例：

from influxdb import InfluxDBClient

client = InfluxDBClient(host='localhost', port=8086)
client.switch_database('wind_data')
results = client.query("SELECT * FROM scada_1s WHERE time > now() - 1h")

4.4 API数据获取

现在很多风机制造商和第三方平台，都提供了RESTful API。比如远景的EnOS、金风的天机平台。通过API，你可以直接拉取数据，不用再跟SCADA系统打交道。

API获取数据，核心就三步：

  1. 认证：通常是Bearer Token或API Key。记得把Token存在环境变量里，别硬编码在代码中。
  2. 构造请求：指定时间范围、风机编号、测点列表。
  3. 解析响应：返回的JSON或CSV，解析后存入数据库。

注意：API通常有频率限制（Rate Limit）。比如每分钟最多请求100次。如果你一次性拉取一年数据，得做好分页和重试机制。我习惯写一个带指数退避的重试函数，遇到429错误自动等待再重试。

一个简单的API数据获取示例：

import requests
import time

def fetch_wind_data(api_url, token, start_time, end_time):
    headers = {"Authorization": f"Bearer {token}"}
    params = {
        "start": start_time,
        "end": end_time,
        "turbine": "WTG-001"
    }
    response = requests.get(api_url, headers=headers, params=params)
    if response.status_code == 200:
        return response.json()
    elif response.status_code == 429:
        time.sleep(5)  # 限流了，等5秒再试
        return fetch_wind_data(api_url, token, start_time, end_time)
    else:
        raise Exception(f"API请求失败: {response.status_code}")

知识体系总览

下面这张图，把本章的核心逻辑串起来了。你可以把它当作数据接入的“作战地图”。

数据采集与接入：核心流程 SCADA系统 CSV/Excel文件 第三方API OPC UA / Modbus Pandas批量导入 RESTful请求 数据存储：MySQL（元数据）+ InfluxDB（时序数据） 数据清洗 → 分析建模 → 可视化

这张图你看懂了吗？数据从左边三个源头进来，经过不同的接入方式，最终汇入数据库。然后才能进行后续的清洗、分析和可视化。每一步都有坑，但每一步也都有解法。

最后说一句：数据接入这件事，看起来简单，但往往是整个项目中最耗时的环节。我建议你从一开始就做好数据字典和接口文档，别等到项目中期再来补。否则，你会像我一样，在深夜对着乱码的CSV文件怀疑人生。

公众号:蓝海资料掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →