基于ElasticSearch实现数据集聚合字段新增及特征自动化构建咨询

Absolutely—you can absolutely add a dedicated field to store rolling averages (or any aggregated calculation) in Elasticsearch, and it’s totally feasible to automate this for your large-scale network log dataset. Let’s walk through the best approach for your bytes_in moving average use case.

Option 1: Use Transforms (Best for Automated, Scalable Storage) #

If you want to persist the moving average results to a field (rather than calculating it on-the-fly during queries), Elasticsearch Transforms are the way to go. They’re designed for automating aggregation jobs and writing results to a target index—perfect for large datasets where you want precomputed features.

Here’s a step-by-step implementation for a 7-day rolling average of bytes_in (adjust the window size as needed):

1. Create the Transform #

Run this DSL to define a transform that groups your logs (e.g., by client_ip if you want per-client averages) and calculates the moving average:

PUT _transform/bytes-in-rolling-avg-transform
{
  "source": {
    "index": "network-logs-*" // Your source network log indices (supports wildcards)
  },
  "dest": {
    "index": "network-logs-with-rolling-avg" // Target index to store results with the new field
  },
  "pivot": {
    "group_by": {
      // Group by client IP (remove this if you want a global rolling average)
      "client_ip": {
        "terms": {
          "field": "client_ip"
        }
      },
      // Group by day to build the rolling window
      "daily_timestamp": {
        "date_histogram": {
          "field": "@timestamp",
          "fixed_interval": "1d",
          "time_zone": "UTC"
        }
      }
    },
    "aggregations": {
      // First calculate daily average bytes_in per group
      "daily_bytes_avg": {
        "avg": {
          "field": "bytes_in"
        }
      },
      // Then compute the rolling average over the last X days (here, 7 days)
      "x_day_rolling_avg_bytes_in": {
        "moving_function": {
          "buckets_path": "daily_bytes_avg",
          "window": 7, // Adjust this to your desired X-day window
          "script": "MovingFunctions.avg(values)"
        }
      }
    }
  },
  "sync": {
    // Keep the transform updated with new logs
    "time": {
      "field": "@timestamp",
      "delay": "1h" // Wait 1 hour to ensure all logs for the day are ingested
    }
  },
  "frequency": "1d" // Run the transform daily to update the rolling average
}

2. Start the Transform #

Once defined, start it to kick off the automation:

POST _transform/bytes-in-rolling-avg-transform/_start

Elasticsearch will now:

• First backfill all historical data to calculate rolling averages for past periods
• Daily update the target index with new rolling average values as fresh logs come in

Option 2: Runtime Fields (For Real-Time Calculation Without Storage) #

If you don’t need to store the moving average (just calculate it when querying), use a Runtime Field. This avoids extra storage but may impact query performance on very large datasets:

Add this runtime field to your log index mapping:

PUT network-logs-*/_mappings
{
  "runtime": {
    "7day_rolling_avg_bytes_in": {
      "type": "double",
      "script": {
        "source": """
          long windowMs = 7 * 24 * 60 * 60 * 1000; // 7 days in milliseconds
          long cutoff = new Date().getTime() - windowMs;
          
          // Query the last 7 days of data for the same client IP
          def hits = _index['network-logs-*'].search(
            query: {
              bool: {
                must: [
                  {term: {client_ip: doc['client_ip'].value}},
                  {range: {@timestamp: {gte: cutoff}}}
                ]
              }
            },
            size: 10000
          ).hits.hits;
          
          double sum = 0;
          int count = 0;
          for (hit in hits) {
            sum += hit['_source']['bytes_in'];
            count++;
          }
          
          emit(count > 0 ? sum / count : null);
        """
      }
    }
  }
}

Key Notes #

• Time Field Requirement: Ensure your logs have a properly mapped date field (like @timestamp)—this is critical for calculating time-based rolling windows.
• Window Adjustment: Change the window value in the transform (or windowMs in the runtime script) to match your desired "past X days" period.
• Scalability: Transforms are optimized for large datasets; they run incrementally and won’t reprocess all data every time. Runtime Fields are better for small to medium datasets where real-time calculation is acceptable.

内容的提问来源于stack exchange，提问作者Louis Hulot