Модель прогнозирования временных рядов с сбоями с XGBOOST и Dask большие наборы данных ⇐ Python

[Anonymous]

Я строю модель прогнозирования временных рядов в Python, чтобы предсказать почасовые кВт -нагрузки для разных типов клиентов в коммунальной компании. Набор данных содержит ~ 81 миллион строк, с данными по почасовой нагрузке для ~ 2300 клиентов в течение 2-4 лет. Тип клиента представлен бинарными столбцами: EV, HP, Solar и Tou. Набор данных имеет следующие переменные: < /p>

Код: Выделить всё

  - read_date: datetime64[us]
- meter: string
- kwh: float64
- city: string
- temperature: float64
- ev: int64
- solar: int64
- hp: int64
- tou: int64
- hour: int32
- day: int32
- month: int32
- year: Int64
- day_of_week: int32
- season: string
- customer_type: string
- hour_sin: float64
- hour_cos: float64
- month_sin: float64
- month_cos: float64
- day_of_week_sin: float64
- day_of_week_cos: float64
- day_sin: float64
- day_cos: float64
- is_holiday: int64
- city_reading: int64
- city_lynnfield: int64
- city_northreading: int64
- city_wilmington: int64
- season_winter: int64
- season_spring: int64
- season_summer: int64
- season_fall: int64

После очистки данных я отказался от следующих функций как из наборов обучения, так и из тестовых наборов: meter , customer_type , сезон , read_date , city , day , month , deam , day_of_week . Моя целевая переменная - почасовая нагрузка кВтч. Я попытался построить модель XGBOOST, используя DASK для распространения, но она продолжает сбой с ошибками, такими как: < /p>

Код: Выделить всё

AssertionError: error
2025-03-31 14:12:26,995 - distributed.nanny - WARNING - Restarting worker
< /code>
Я работаю над локальной машиной с 128 ГБ оперативной памяти и процессором Intel i7-14700K 3,40 ГГц. Я ищу рекомендации о том, как справиться с прогнозированием временных рядов с помощью этого большого набора данных и как избежать сбоев при использовании DASK для распространения.  Вот мой пример кода: < /p>
# Import necessary libraries
import numpy as np
import dask.dataframe as dd
import dask.array as da
import xgboost as xgb
from dask.distributed import Client
from dask.diagnostics import ProgressBar
from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score
import warnings
import matplotlib.pyplot as plt
from tqdm import tqdm

# Load the data using Dask (efficient for large Parquet files)
some_feats_dd = dd.read_parquet("pre_ml_some_features.parquet")

# Rename DataFrame
df_processed = some_feats_dd

# Filter the data based on the read_date for training and testing
df_train = df_processed[df_processed["year"] < 2025]  # Keep rows before 2025
df_test = df_processed[df_processed["year"] == 2025]  # Keep rows from 2025 onwards

# Exclude columns and prepare features and target variables for training
exclude_cols = ["kwh", "meter", "customer_type", "season", "read_date", "city",
"day", "month", "hour", "day_of_week"]

# Prepare training features (X) and target variable (y)
X_train = df_train.drop(columns=exclude_cols)
y_train = df_train["kwh"]

# Compute total lengths and ensure exact 3 chunks
train_size = len(y_train.compute())
test_size = len(df_test)  # No need to compute, Dask can infer

# Convert y_train and y_test to Dask arrays with forced 3 chunks
y_train = da.from_array(y_train.compute(), chunks=(train_size // 3,))
y_test = da.from_array(df_test["kwh"].compute(), chunks=(test_size // 2,))

# Ensure partitions match for X_train and X_test
X_train = X_train.repartition(npartitions=3)
X_test = X_test.repartition(npartitions=3)

# Start Dask client for parallel processing
client = Client()

# Print the Dask dashboard URL
print(f"Dask dashboard is available at: {client.dashboard_link}")

# Use DaskDMatrix from xgboost.dask
dask_train_data = xgb.dask.DaskDMatrix(client, X_train, y_train)

# Set up parameters for XGBoost
params = {
'objective': 'reg:squarederror',  # Regression task
'eval_metric': 'rmse',
'tree_method': 'hist',  # Use histogram-based method for faster training
'verbosity': 1,  # Enables basic logging
}

# Initialize Dask-XGBoost model
dask_gbr = xgb.dask.DaskXGBRegressor(**params)

# Train the model using Dask (this will automatically parallelize)
with ProgressBar():  # Shows progress during training
dask_gbr.fit(dask_train_data)

Спасибо!

Подробнее здесь: https://stackoverflow.com/questions/795 ... s-crashing