Samouczek: tworzenie modeli opartych na partycjach w języku R w programie SQL Server

Dotyczy: SQL Server 2016 (13.x) i nowsze wersje

W programie SQL Server 2019 modelowanie oparte na partycjach to możliwość tworzenia i trenowania modeli na podstawie partycjonowanych danych. W przypadku danych stratyfikowanych, które naturalnie segmentują się w danym schemacie klasyfikacji — takich jak regiony geograficzne, daty i godziny, wiek lub płeć — można wykonać skrypt na całym zbiorze danych, z możliwością modelowania, trenowania i oceny partycji, które pozostają nieruszone podczas wszystkich tych operacji.

Modelowanie oparte na partycjach jest włączone za pomocą dwóch nowych parametrów w sp_execute_external_script:

• input_data_1_partition_by_columns, który określa kolumnę, według której następuje partycjonowanie.
• input_data_1_order_by_columns określa, według których kolumn ma nastąpić sortowanie.

W tym samouczku nauczysz się modelowania opartego na partycjach przy użyciu klasycznych danych przykładowych taksówek w Nowym Jorku i skryptu języka R. Kolumna podziału to metoda płatności.

Warunki wstępne

Aby ukończyć ten samouczek, musisz mieć następujące elementy:

• Wystarczająca ilość zasobów systemowych. Zestaw danych jest duży, a operacje szkoleniowe intensywnie korzystają z zasobów. Jeśli to możliwe, użyj systemu z co najmniej 8 GB pamięci RAM. Alternatywnie można użyć mniejszych zestawów danych, aby obejść ograniczenia zasobów. Instrukcje dotyczące zmniejszania zestawu danych są wbudowane.

• Narzędzie do wykonywania zapytań T-SQL, takie jak SQL Server Management Studio (SSMS).

• NYCTaxi_Sample.bak, które można pobrać i przywrócić do lokalnego wystąpienia SQL Server. Rozmiar pliku wynosi około 90 MB.

• Wystąpienie aparatu bazy danych programu SQL Server 2019 z usługami Machine Learning Services i integracją języka R.

• W tym samouczku jest używane połączenie sprzężenia zwrotnego z programem SQL Server ze skryptu języka R za pośrednictwemODBC. W związku z tym należy utworzyć konto dla SQLRUserGroup.

• Sprawdź dostępność pakietów języka R, zwracając dobrze sformatowaną listę wszystkich pakietów języka R aktualnie zainstalowanych z instancją silnika bazy danych.

EXECUTE sp_execute_external_script
  @language=N'R',
  @script = N'str(OutputDataSet);
  packagematrix <- installed.packages();
  Name <- packagematrix[,1];
  Version <- packagematrix[,3];
  OutputDataSet <- data.frame(Name, Version);',
  @input_data_1 = N''
WITH RESULT SETS ((PackageName nvarchar(250), PackageVersion nvarchar(max) ))

Nawiązywanie połączenia z bazą danych

Uruchom program SSMS i połącz się z wystąpieniem silnika bazy danych. W eksploratorze obiektów sprawdź, czy istnieje bazy danych NYCTaxi_Sample.

Utwórz CalculateDistance

Baza danych demonstracyjna zawiera funkcję skalarną do obliczania odległości, ale nasza procedura składowana działa lepiej z funkcją o wartości tabeli. Później uruchom następujący skrypt, aby utworzyć funkcję CalculateDistance używaną w kroku trenowania .

Aby potwierdzić utworzenie funkcji, w eksploratorze obiektów sprawdź \Programmability\Functions\Table-valued Functions w bazie danych NYCTaxi_Sample.

USE NYCTaxi_sample
GO

SET ANSI_NULLS ON
GO

SET QUOTED_IDENTIFIER ON
GO

CREATE FUNCTION [dbo].[CalculateDistance] (
    @Lat1 FLOAT
    ,@Long1 FLOAT
    ,@Lat2 FLOAT
    ,@Long2 FLOAT
    )
    -- User-defined function calculates the direct distance between two geographical coordinates.
RETURNS TABLE
AS
RETURN

SELECT COALESCE(3958.75 * ATAN(SQRT(1 - POWER(t.distance, 2)) / nullif(t.distance, 0)), 0) AS direct_distance
FROM (
    VALUES (CAST((SIN(@Lat1 / 57.2958) * SIN(@Lat2 / 57.2958)) + (COS(@Lat1 / 57.2958) * COS(@Lat2 / 57.2958) * COS((@Long2 / 57.2958) - (@Long1 / 57.2958))) AS DECIMAL(28, 10)))
    ) AS t(distance)
GO

Zdefiniuj procedurę tworzenia i trenowania modeli dla podziału na partycje

Ten samouczek opakowuje skrypt języka R w procedurze składowanej. W tym kroku utworzysz procedurę składowaną, która używa języka R do tworzenia wejściowego zestawu danych, tworzenia modelu klasyfikacji do przewidywania wyników porad, a następnie przechowywania modelu w bazie danych.

Wśród danych wejściowych parametrów używanych przez ten skrypt zobaczysz input_data_1_partition_by_columns i input_data_1_order_by_columns. Pamiętaj, że te parametry są mechanizmem, za pomocą którego odbywa się modelowanie partycjonowane. Parametry są przekazywane jako dane wejściowe do sp_execute_external_script przetwarzania partycji za pomocą skryptu zewnętrznego wykonywanego raz dla każdej partycji.

W przypadku tej procedury składowanej użyj przetwarzania równoległego, aby skrócić czas wykonania.

Po uruchomieniu tego skryptu, w Eksploratorze obiektów , powinieneś zobaczyć train_rxLogIt_per_partition w \Programmability\Stored Procedures, pod bazą danych NYCTaxi_Sample. Powinna zostać również wyświetlona nowa tabela używana do przechowywania modeli: dbo.nyctaxi_models.

USE NYCTaxi_Sample
GO

CREATE
    OR

ALTER PROCEDURE [dbo].[train_rxLogIt_per_partition] (@input_query NVARCHAR(max))
AS
BEGIN
    DECLARE @start DATETIME2 = SYSDATETIME()
        ,@model_generation_duration FLOAT
        ,@model VARBINARY(max)
        ,@instance_name NVARCHAR(100) = @@SERVERNAME
        ,@database_name NVARCHAR(128) = db_name();

    EXEC sp_execute_external_script @language = N'R'
        ,@script =
        N'
    
    # Make sure InputDataSet is not empty. In parallel mode, if one thread gets zero data, an error occurs
    if (nrow(InputDataSet) > 0) {
    # Define the connection string
    connStr <- paste("Driver=SQL Server;Server=", instance_name, ";Database=", database_name, ";Trusted_Connection=true;", sep="");
    
    # build classification model to predict a tip outcome
    duration <- system.time(logitObj <- rxLogit(tipped ~ passenger_count + trip_distance + trip_time_in_secs + direct_distance, data = InputDataSet))[3];

    # First, serialize a model to and put it into a database table
    modelbin <- as.raw(serialize(logitObj, NULL));

    # Create the data source. To reduce data size, add rowsPerRead=500000 to cut the dataset by half.
    ds <- RxOdbcData(table="ml_models", connectionString=connStr);

    # Store the model in the database
    model_name <- paste0("nyctaxi.", InputDataSet[1,]$payment_type);
    
    rxWriteObject(ds, model_name, modelbin, version = "v1",
    keyName = "model_name", valueName = "model_object", versionName = "model_version", overwrite = TRUE, serialize = FALSE);
    }
    
    '
        ,@input_data_1 = @input_query
        ,@input_data_1_partition_by_columns = N'payment_type'
        ,@input_data_1_order_by_columns = N'passenger_count'
        ,@parallel = 1
        ,@params = N'@instance_name nvarchar(100), @database_name nvarchar(128)'
        ,@instance_name = @instance_name
        ,@database_name = @database_name
    WITH RESULT SETS NONE
END;
GO

Wykonanie równoległe

Zwróć uwagę, że dane wejściowe sp_execute_external_script obejmują @parallel=1, używane do włączania przetwarzania równoległego. W przeciwieństwie do poprzednich wersji, począwszy od programu SQL Server 2019, ustawienie @parallel=1 zapewnia silniejszą wskazówkę dla optymalizatora zapytań, dzięki czemu wykonywanie równoległe będzie znacznie bardziej prawdopodobne.

Domyślnie optymalizator zapytań ma tendencję do działania pod @parallel=1 na tabelach mających więcej niż 256 wierszy, ale można to zrobić jawnie, ustawiając @parallel=1, jak pokazano w tym skrypcie.

Uruchamianie procedury i trenowanie modelu

W tej sekcji skrypt szkoli utworzony i zapisany w poprzednim kroku model. W poniższych przykładach pokazano dwa podejścia do trenowania modelu: przy użyciu całego zestawu danych lub częściowych danych.

Spodziewaj się, że ten krok zajmie trochę czasu. Trenowanie jest intensywnie obciążane obliczeniami, trwa wiele minut. Jeśli zasoby systemowe, zwłaszcza pamięć, nie są wystarczające do załadowania, użyj podzestawu danych. Drugi przykład zawiera składnię.

--Example 1: train on entire dataset
EXEC train_rxLogIt_per_partition N'
SELECT payment_type, tipped, passenger_count, trip_time_in_secs, trip_distance, d.direct_distance
  FROM dbo.nyctaxi_sample CROSS APPLY [CalculateDistance](pickup_latitude, pickup_longitude,  dropoff_latitude, dropoff_longitude) as d
';
GO

--Example 2: Train on 20 percent of the dataset to expedite processing.
EXEC train_rxLogIt_per_partition N'
  SELECT tipped, payment_type, passenger_count, trip_time_in_secs, trip_distance, d.direct_distance
  FROM dbo.nyctaxi_sample TABLESAMPLE (20 PERCENT) REPEATABLE (98074)
  CROSS APPLY [CalculateDistance](pickup_latitude, pickup_longitude,  dropoff_latitude, dropoff_longitude) as d
';
GO

Sprawdzanie wyników

Wynikiem tabeli modeli powinno być pięć różnych modeli na podstawie pięciu partycji podzielonych na segmenty według pięciu typów płatności. Modele znajdują się w ml_models źródle danych.

SELECT *
FROM ml_models

Definiowanie procedury przewidywania wyników

Do oceniania można użyć tych samych parametrów. Poniższy przykład zawiera skrypt języka R, który będzie oceniać przy użyciu poprawnego modelu dla partycji, która jest obecnie przetwarzana.

Tak jak wcześniej, utwórz procedurę przechowywaną, aby opakować kod języka R.

USE NYCTaxi_Sample
GO

-- Stored procedure that scores per partition.
-- Depending on the partition being processed, a model specific to that partition will be used
CREATE
    OR

ALTER PROCEDURE [dbo].[predict_per_partition]
AS
BEGIN
    DECLARE @predict_duration FLOAT
        ,@instance_name NVARCHAR(100) = @@SERVERNAME
        ,@database_name NVARCHAR(128) = db_name()
        ,@input_query NVARCHAR(max);

    SET @input_query = 'SELECT tipped, passenger_count, trip_time_in_secs, trip_distance, d.direct_distance, payment_type
                          FROM dbo.nyctaxi_sample TABLESAMPLE (1 PERCENT) REPEATABLE (98074)
                          CROSS APPLY [CalculateDistance](pickup_latitude, pickup_longitude,  dropoff_latitude, dropoff_longitude) as d'

    EXEC sp_execute_external_script @language = N'R'
        ,@script =
        N'
    
    if (nrow(InputDataSet) > 0) {

    #Get the partition that is currently being processed
    current_partition <- InputDataSet[1,]$payment_type;

    #Create the SQL query to select the right model
    query_getModel <- paste0("select model_object from ml_models where model_name = ", "''", "nyctaxi.",InputDataSet[1,]$payment_type,"''", ";")
    

    # Define the connection string
    connStr <- paste("Driver=SQL Server;Server=", instance_name, ";Database=", database_name, ";Trusted_Connection=true;", sep="");
        
    #Define data source to use for getting the model
    ds <- RxOdbcData(sqlQuery = query_getModel, connectionString = connStr)

    # Load the model
    modelbin <- rxReadObject(ds, deserialize = FALSE)
    # unserialize model
    logitObj <- unserialize(modelbin);

    # predict tipped or not based on model
    predictions <- rxPredict(logitObj, data = InputDataSet, overwrite = TRUE, type = "response", writeModelVars = TRUE
        , extraVarsToWrite = c("payment_type"));
    OutputDataSet <- predictions
    
    } else {
        OutputDataSet <- data.frame(integer(), InputDataSet[,]);
    }
    '
        ,@input_data_1 = @input_query
        ,@parallel = 1
        ,@input_data_1_partition_by_columns = N'payment_type'
        ,@params = N'@instance_name nvarchar(100), @database_name nvarchar(128)'
        ,@instance_name = @instance_name
        ,@database_name = @database_name
    WITH RESULT SETS((
                tipped_Pred INT
                ,payment_type VARCHAR(5)
                ,tipped INT
                ,passenger_count INT
                ,trip_distance FLOAT
                ,trip_time_in_secs INT
                ,direct_distance FLOAT
                ));
END;
GO

Tworzenie tabeli do przechowywania przewidywań

CREATE TABLE prediction_results (
    tipped_Pred INT
    ,payment_type VARCHAR(5)
    ,tipped INT
    ,passenger_count INT
    ,trip_distance FLOAT
    ,trip_time_in_secs INT
    ,direct_distance FLOAT
    );

TRUNCATE TABLE prediction_results
GO

Uruchamianie procedury i zapisywanie przewidywań

INSERT INTO prediction_results (
    tipped_Pred
    ,payment_type
    ,tipped
    ,passenger_count
    ,trip_distance
    ,trip_time_in_secs
    ,direct_distance
    )
EXECUTE [predict_per_partition]
GO

Wyświetlanie przewidywań

Ponieważ przewidywania są przechowywane, możesz uruchomić proste zapytanie, aby zwrócić zestaw wyników.

SELECT *
FROM prediction_results;

Następne kroki

• W tym samouczku użyto sp_execute_external_script do iterowania operacji na danych partycjonowanych. Aby bliżej przyjrzeć się wywoływaniu skryptów zewnętrznych w procedurach składowanych i używaniu funkcji RevoScaleR, przejdź do następującego samouczka.