Apache Doris 在 IOT 场景的实践与落地方案（内附案例）

DorisLogstash设备DorisLogstash设备发送原始日志解析JSON/过滤Stream Load写入确认状态反馈

input {
  file {
    path => "/path/to/log/*.log"
    type => "iot_log"
  }
}

filter {
  json {
    source => "message"
  }
  date {
    match => ["timestamp", "yyyy-MM-dd HH:mm:ss"]
    target => "@timestamp"
  }
}

output {
  http {
    url => "http://doris_host:8030/api/db/table/_stream_load"
    http_method => "PUT"
    format => "json"
  }
}

sensors:
  -name:temperature
    type:analog
    pin:
    interval:100ms
-name:vibration
    type:digital
    pin:
    trigger:edge

output:
type:doris
endpoint:doris-gateway:8030
batch_size:
flush_interval: 1s

def transform(payload):
    return {
        "ts": datetime.now().isoformat(),
        "device_id": payload['device_id'],
        "metrics": {
            "temp": round(payload['temperature'], ),
            "vib": payload['vibration'] * 0.1
        }
    }

CREATE ROUTINE LOAD db.iot_kafka_load ON iot_metrics
PROPERTIES
(
    "desired_concurrent_number" = "4", -- 并行度
    "max_batch_interval" = "20",       -- 最大消费间隔
    "max_batch_rows" = "200000",       -- 单批最大行数
    "strict_mode" = "false"
)
FROM KAFKA
(
    "kafka_broker_list" = "broker1:9092,broker2:9092",
    "kafka_topic" = "iot-raw-data",
    "property.security.protocol" = "sasl_ssl",
    "property.sasl.mechanism" = "PLAIN",
    "property.sasl.username" = "doris",
    "property.sasl.password" = "encrypted_password"
);

"jsonpaths" = "['$.device.id', '$.ts', '$.data.temperature', '$.data.vibration']",
"json_root" = "$.payload",
"strip_outer_array" = "true"

"max_error_number" = "1000",          -- 最大容错数
"load_to_single_tablet" = "false",     -- 分散写入压力
"format" = "json",
"num_as_string" = "true"              -- 处理大整数

def stream_load(data):
    headers = {
        "Authorization": "Basic " + base64.b64encode(f"{username}:{password}".encode()).decode(),
        "format": "json",
        "strip_outer_array": "true"
    }
    
    response = requests.put(
        url='http://fe_host:8030/api/db/table/_stream_load',
        headers=headers,
        data=json.dumps(data)
    )
    
    if response.json()['Status'] != 'Success':
        raise Exception("Load failed: " + response.text)

DorisSink<DeviceData> sink = DorisSink.<DeviceData>builder()
    .setFenodes("fe_host:8030")
    .setUsername("user")
    .setPassword("pass")
    .setTableIdentifier("db.table")
    .setSerializer(new DeviceDataSerializer())
    .build();

env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "Kafka Source")
   .process(new MetricParser())
   .sinkTo(sink);

CREATE TABLE iot_metrics
(
    `timestamp` DATETIME,
    `device_id` VARCHAR(),
    `metrics` JSON,
    `tags` BITMAP
)
PARTITION BY RANGE(`timestamp`)
(
    PARTITION p1 VALUES LESS THAN ('2024-01-01'),
    PARTITION p2 VALUES LESS THAN ('2024-02-01')
)
DISTRIBUTED BY HASH(`device_id`) BUCKETS ;

CREATE MATERIALIZED VIEW mv_hourly_agg
REFRESH ASYNC
AS SELECT 
    device_id,
    DATE_TRUNC(timestamp, 'HOUR') as hour,
    COUNT(*) as count,
    AVG(metrics.temperature) as avg_temp
FROM iot_metrics
GROUP BY device_id, DATE_TRUNC(timestamp, 'HOUR');

-- 实时车辆位置追踪
SELECT
    device_id,
    metrics.longitude,
    metrics.latitude,
    metrics.speed,
    NOW() -timestampAS latency 
FROM iot_metrics
WHERE
    device_id ='VIN_1001'
    AND timestamp >= NOW() - INTERVAL SECOND
ORDERBY timestamp DESC
LIMIT ;

-- 事故前后5分钟数据回溯
WITH accident_time AS (
SELECT MIN(timestamp) AS accident_ts
FROM iot_metrics
WHERE device_id ='VIN_2002'AND metrics.collision =true
)
SELECT *
FROM iot_metrics
WHERE device_id ='VIN_2002'
AND timestamp BETWEEN
      (SELECT accident_ts - INTERVAL MINUTEFROM accident_time)
      AND
      (SELECT accident_ts + INTERVAL MINUTEFROM accident_time);

-- 设备健康特征提取
CREATE MATERIALIZED VIEW device_health_features
REFRESH ASYNC EVERY HOUR
AS
SELECT
    device_id,
    DATE_TRUNC('hour', timestamp) AS time_window,
    AVG(metrics.temperature) AS avg_temp,
    STDDEV(metrics.vibration) AS vib_stddev,
    COUNT_IF(metrics.error_code ISNOT NULL) AS error_count,
    MAX(metrics.pressure) -MIN(metrics.pressure) AS pressure_range
FROM iot_metrics
GROUPBY device_id, time_window;

from doris.flight import connect

# 建立Arrow Flight连接
conn = connect(host='doris-fe', port=, user='ml', password='ml@2024')

# 批量获取训练数据
reader = conn.execute("SELECT * FROM device_health_features")
batch = reader.read_all()

# 转换为Pandas DataFrame
df = batch.to_pandas()

# 训练XGBoost模型
model = xgb.train(params, xgb.DMatrix(df[features], label=df['label']))

# 将模型特征回写Doris
model_features = pd.DataFrame({
    'model_version': ['v1.2'],
    'features': [','.join(features)],
    'importance': [model.feature_importances_.tolist()]
})

conn.upload("model_features", model_features)

消息中心计算引擎地理围栏库规则引擎Doris车辆消息中心计算引擎地理围栏库规则引擎Doris车辆上报位置(1s/次)触发围栏检查获取围栏定义返回围栏列表空间关系计算越界结果发送告警

-- 动态拥堵指数计算
WITH grid_traffic AS (
SELECT
    FLOOR(lng*) AS grid_x,
    FLOOR(lat*) AS grid_y,
    COUNT(*) AS vehicle_count,
    AVG(speed) AS avg_speed
FROM realtime_location
WHERE ts >= NOW() -INTERVALMINUTE
GROUPBY ,
)
SELECT
  grid_x, grid_y,
  vehicle_count * (- avg_speed/) AS congestion_index
FROM grid_traffic
WHERE congestion_index >0.7;

-- 急加速行为识别
SELECT 
  device_id,
  COUNT_IF(accel>0.3) AS rapid_accel_count 
FROM driving_behavior
WHERE time BETWEEN '2024-03-01' AND '2024-03-31'
GROUP BY ;

-- 快速筛查异常设备
SELECT device_id 
FROM sensor_anomaly_index 
WHERE bitmap_contains(anomaly_types, ); -- 205为刹车传感器异常编码

-- 多源传感器特征融合
CREATE MATERIALIZED VIEW gearbox_health REFRESH COMPLETE ON SCHEDULE EVERY 1 hour AS
SELECT
  device_id,
  time_bucket('10 minutes', ts) AS time_window,
AVG(vibration_x) +0.3*AVG(vibration_y) AS composite_vib,
  WAVELET(vibration, ) AS wavelet_coeff,
  FFT(vibration, )->'harmonics'AS freq_domain
FROM turbine_sensors
GROUP BY ,;

-- 参数相关性分析
SELECT 
  CORR(chamber_pressure, defect_rate) AS pressure_corr,
  CORR(plasma_power, defect_rate) AS power_corr
FROM production_batches
WHERE process_step = 'etching';

-- 工序能耗异常检测
WITH current_energy AS (
SELECT process_line, SUM(power) AS total 
FROM energy_stream 
WHERE ts >= NOW() -INTERVALMINUTE
GROUP BY 
),
historical_avg AS (
SELECT process_line, AVG(power) AS avg_power
FROM energy_history
WHERE time_range ='same_day_last_week'
GROUP BY 
)
SELECT a.process_line, a.total, b.avg_power
FROM current_energy a
JOIN historical_avg b ON a.process_line = b.process_line
WHERE a.total > b.avg_power *1.2;