Przewodnik programowania U-SQL — UDT i UDAGG

  • Artykuł

Używanie typów zdefiniowanych przez użytkownika: UDT

Typy zdefiniowane przez użytkownika (UDT) to kolejna funkcja programowania języka U-SQL. U-SQL UDT działa jak zwykły typ zdefiniowany przez użytkownika języka C#. C# to silnie typizowane język, który umożliwia korzystanie z wbudowanych i niestandardowych typów zdefiniowanych przez użytkownika.

Język U-SQL nie może niejawnie serializować ani deserializować dowolnych tras zdefiniowanych przez użytkownika, gdy funkcja UDT jest przekazywana między wierzchołkami w zestawach wierszy. Oznacza to, że użytkownik musi podać jawny formater przy użyciu interfejsu IFormatter. Zapewnia to język U-SQL z serializacji i deserializacji metod udT.

Uwaga

Wbudowane moduły wyodrębniające i wyjściowe języka U-SQL nie mogą obecnie serializować ani deserializować danych UDT do lub z plików nawet przy użyciu zestawu IFormatter. Dlatego podczas zapisywania danych UDT w pliku za pomocą instrukcji OUTPUT lub odczytywania ich za pomocą modułu wyodrębniającego należy przekazać je jako ciąg lub tablicę bajtów. Następnie należy jawnie wywołać kod serializacji i deserializacji (czyli metody ToString() udT. Z drugiej strony moduły wyodrębniające i wyjściowe zdefiniowane przez użytkownika mogą odczytywać i zapisywać trasy zdefiniowane przez użytkownika.

Jeśli spróbujemy użyć funkcji UDT w funkcji EXTRACTOR lub OUTPUTTER (z poprzedniej funkcji SELECT), jak pokazano poniżej:

@rs1 =
    SELECT
        MyNameSpace.Myfunction_Returning_UDT(filed1) AS myfield
    FROM @rs0;

OUTPUT @rs1
    TO @output_file
    USING Outputters.Text();

Występuje następujący błąd:

Error	1	E_CSC_USER_INVALIDTYPEINOUTPUTTER: Outputters.Text was used to output column myfield of type
MyNameSpace.Myfunction_Returning_UDT.

Description:

Outputters.Text only supports built-in types.

Resolution:

Implement a custom outputter that knows how to serialize this type, or call a serialization method on the type in
the preceding SELECT.	C:\Users\sergeypu\Documents\Visual Studio 2013\Projects\USQL-Programmability\
USQL-Programmability\Types.usql	52	1	USQL-Programmability

Aby pracować z funkcją UDT w module wyjściowym, musimy serializować go do ciągu za pomocą metody ToString() lub utworzyć niestandardowy moduł wyjściowy.

Obecnie w funkcji GROUP BY nie można używać zdefiniowanych tras zdefiniowanych przez użytkownika. Jeśli funkcja UDT jest używana w funkcji GROUP BY, zgłaszany jest następujący błąd:

Error	1	E_CSC_USER_INVALIDTYPEINCLAUSE: GROUP BY doesn't support type MyNameSpace.Myfunction_Returning_UDT
for column myfield

Description:

GROUP BY doesn't support UDT or Complex types.

Resolution:

Add a SELECT statement where you can project a scalar column that you want to use with GROUP BY.
C:\Users\sergeypu\Documents\Visual Studio 2013\Projects\USQL-Programmability\USQL-Programmability\Types.usql
62	5	USQL-Programmability

Aby zdefiniować udT, musimy:

  1. Dodaj następujące przestrzenie nazw:
using Microsoft.Analytics.Interfaces
using System.IO;

  1. Dodaj Microsoft.Analytics.Interfaceselement , który jest wymagany dla interfejsów UDT. System.IO Ponadto może być konieczne zdefiniowanie interfejsu IFormatter.

  2. Zdefiniuj typ zdefiniowany przez użytkownika za pomocą atrybutu SqlUserDefinedType.

SqlUserDefinedType służy do oznaczania definicji typu w zestawie jako typu zdefiniowanego przez użytkownika (UDT) w języku U-SQL. Właściwości atrybutu odzwierciedlają cechy fizyczne udT. Tej klasy nie można dziedziczyć.

SqlUserDefinedType jest atrybutem wymaganym dla definicji UDT.

Konstruktor klasy:

  • SqlUserDefinedTypeAttribute (formater typów)

  • Formater typów: wymagany parametr do zdefiniowania formatowania UDT — w szczególności typ interfejsu IFormatter musi zostać przekazany tutaj.

[SqlUserDefinedType(typeof(MyTypeFormatter))]
public class MyType
{ … }
  • Typowa funkcja UDT wymaga również definicji interfejsu IFormatter, jak pokazano w poniższym przykładzie:
public class MyTypeFormatter : IFormatter<MyType>
{
    public void Serialize(MyType instance, IColumnWriter writer, ISerializationContext context)
    { … }

    public MyType Deserialize(IColumnReader reader, ISerializationContext context)
    { … }
}

Interfejs IFormatter serializuje i deserializuje graf obiektu z typem głównym typem typeparamref <name="T".>

<typeparam name="T">Typ główny grafu obiektu do serializacji i deserializacji.

  • Deserializowanie: deserializuje dane w podanym strumieniu i odtwarza graf obiektów.

  • Serializowanie: serializuje obiekt lub graf obiektów z danym elementem głównym dla podanego strumienia.

MyType wystąpienie: wystąpienie typu. IColumnWriter zapis/ IColumnReader czytelnik: podstawowy strumień kolumny. ISerializationContext context: wyliczenie definiujące zestaw flag określający kontekst źródłowy lub docelowy strumienia podczas serializacji.

  • Pośredni: określa, że kontekst źródłowy lub docelowy nie jest utrwalonego magazynu.

  • Trwałość: określa, że kontekst źródłowy lub docelowy jest utrwalonego magazynu.

Jako zwykły typ języka C# definicja U-SQL UDT może zawierać przesłonięcia dla operatorów, takich jak +/==/!=. Może również zawierać metody statyczne. Jeśli na przykład użyjemy tego udT jako parametru funkcji agregującej U-SQL MIN, musimy zdefiniować < zastąpienie operatora.

Wcześniej w tym przewodniku przedstawiono przykład identyfikacji okresu obrachunkowego z określonej daty w formacie Qn:Pn (Q1:P10). W poniższym przykładzie pokazano, jak zdefiniować typ niestandardowy dla wartości okresu obrachunkowego.

Poniżej znajduje się przykład sekcji kodu z niestandardowym interfejsem UDT i IFormatter:

[SqlUserDefinedType(typeof(FiscalPeriodFormatter))]
public struct FiscalPeriod
{
    public int Quarter { get; private set; }

    public int Month { get; private set; }

    public FiscalPeriod(int quarter, int month):this()
    {
        this.Quarter = quarter;
        this.Month = month;
    }

    public override bool Equals(object obj)
    {
        if (ReferenceEquals(null, obj))
        {
            return false;
        }

        return obj is FiscalPeriod && Equals((FiscalPeriod)obj);
    }

    public bool Equals(FiscalPeriod other)
    {
return this.Quarter.Equals(other.Quarter) && this.Month.Equals(other.Month);
    }

    public bool GreaterThan(FiscalPeriod other)
    {
return this.Quarter.CompareTo(other.Quarter) > 0 || this.Month.CompareTo(other.Month) > 0;
    }

    public bool LessThan(FiscalPeriod other)
    {
return this.Quarter.CompareTo(other.Quarter) < 0 || this.Month.CompareTo(other.Month) < 0;
    }

    public override int GetHashCode()
    {
        unchecked
        {
            return (this.Quarter.GetHashCode() * 397) ^ this.Month.GetHashCode();
        }
    }

    public static FiscalPeriod operator +(FiscalPeriod c1, FiscalPeriod c2)
    {
return new FiscalPeriod((c1.Quarter + c2.Quarter) > 4 ? (c1.Quarter + c2.Quarter)-4 : (c1.Quarter + c2.Quarter), (c1.Month + c2.Month) > 12 ? (c1.Month + c2.Month) - 12 : (c1.Month + c2.Month));
    }

    public static bool operator ==(FiscalPeriod c1, FiscalPeriod c2)
    {
        return c1.Equals(c2);
    }

    public static bool operator !=(FiscalPeriod c1, FiscalPeriod c2)
    {
        return !c1.Equals(c2);
    }
    public static bool operator >(FiscalPeriod c1, FiscalPeriod c2)
    {
        return c1.GreaterThan(c2);
    }
    public static bool operator <(FiscalPeriod c1, FiscalPeriod c2)
    {
        return c1.LessThan(c2);
    }
    public override string ToString()
    {
        return (String.Format("Q{0}:P{1}", this.Quarter, this.Month));
    }

}

public class FiscalPeriodFormatter : IFormatter<FiscalPeriod>
{
    public void Serialize(FiscalPeriod instance, IColumnWriter writer, ISerializationContext context)
    {
        using (var binaryWriter = new BinaryWriter(writer.BaseStream))
        {
            binaryWriter.Write(instance.Quarter);
            binaryWriter.Write(instance.Month);
            binaryWriter.Flush();
        }
    }

    public FiscalPeriod Deserialize(IColumnReader reader, ISerializationContext context)
    {
        using (var binaryReader = new BinaryReader(reader.BaseStream))
        {
var result = new FiscalPeriod(binaryReader.ReadInt16(), binaryReader.ReadInt16());
            return result;
        }
    }
}

Zdefiniowany typ zawiera dwie liczby: kwartał i miesiąc. Operatory ==/!=/>/< i metoda ToString() statyczna są zdefiniowane w tym miejscu.

Jak wspomniano wcześniej, funkcja UDT może być używana w wyrażeniach SELECT, ale nie może być używana w funkcji OUTPUTTER/EXTRACTOR bez niestandardowej serializacji. Musi być serializowany jako ciąg z ToString() niestandardowym modułem OUTPUTTER/EXTRACTOR lub używany z nim.

Teraz omówimy użycie funkcji UDT. W sekcji kodu zmieniliśmy naszą funkcję GetFiscalPeriod na następującą:

public static FiscalPeriod GetFiscalPeriodWithCustomType(DateTime dt)
{
    int FiscalMonth = 0;
    if (dt.Month < 7)
    {
        FiscalMonth = dt.Month + 6;
    }
    else
    {
        FiscalMonth = dt.Month - 6;
    }

    int FiscalQuarter = 0;
    if (FiscalMonth >= 1 && FiscalMonth <= 3)
    {
        FiscalQuarter = 1;
    }
    if (FiscalMonth >= 4 && FiscalMonth <= 6)
    {
        FiscalQuarter = 2;
    }
    if (FiscalMonth >= 7 && FiscalMonth <= 9)
    {
        FiscalQuarter = 3;
    }
    if (FiscalMonth >= 10 && FiscalMonth <= 12)
    {
        FiscalQuarter = 4;
    }

    return new FiscalPeriod(FiscalQuarter, FiscalMonth);
}

Jak widać, zwraca wartość typu FiscalPeriod.

W tym miejscu przedstawiono przykład sposobu dalszego używania go w skry skry skryptie podstawowym U-SQL. W tym przykładzie pokazano różne formy wywołania UDT ze skryptu U-SQL.

DECLARE @input_file string = @"c:\work\cosmos\usql-programmability\input_file.tsv";
DECLARE @output_file string = @"c:\work\cosmos\usql-programmability\output_file.tsv";

@rs0 =
    EXTRACT
        guid string,
        dt DateTime,
        user String,
        des String
    FROM @input_file USING Extractors.Tsv();

@rs1 =
    SELECT
        guid AS start_id,
        dt,
        DateTime.Now.ToString("M/d/yyyy") AS Nowdate,
        USQL_Programmability.CustomFunctions.GetFiscalPeriodWithCustomType(dt).Quarter AS fiscalquarter,
        USQL_Programmability.CustomFunctions.GetFiscalPeriodWithCustomType(dt).Month AS fiscalmonth,
        USQL_Programmability.CustomFunctions.GetFiscalPeriodWithCustomType(dt) + new USQL_Programmability.CustomFunctions.FiscalPeriod(1,7) AS fiscalperiod_adjusted,
        user,
        des
    FROM @rs0;

@rs2 =
    SELECT
        start_id,
        dt,
        DateTime.Now.ToString("M/d/yyyy") AS Nowdate,
        fiscalquarter,
        fiscalmonth,
        USQL_Programmability.CustomFunctions.GetFiscalPeriodWithCustomType(dt).ToString() AS fiscalperiod,

           // This user-defined type was created in the prior SELECT.  Passing the UDT to this subsequent SELECT would have failed if the UDT was not annotated with an IFormatter.
           fiscalperiod_adjusted.ToString() AS fiscalperiod_adjusted,
           user,
           des
    FROM @rs1;

OUTPUT @rs2
    TO @output_file
    USING Outputters.Text();

Oto przykład pełnej sekcji kodu:

using Microsoft.Analytics.Interfaces;
using Microsoft.Analytics.Types.Sql;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.IO;

namespace USQL_Programmability
{
    public class CustomFunctions
    {
        static public DateTime? ToDateTime(string dt)
        {
            DateTime dtValue;

            if (!DateTime.TryParse(dt, out dtValue))
                return Convert.ToDateTime(dt);
            else
                return null;
        }

        public static FiscalPeriod GetFiscalPeriodWithCustomType(DateTime dt)
        {
            int FiscalMonth = 0;
            if (dt.Month < 7)
            {
                FiscalMonth = dt.Month + 6;
            }
            else
            {
                FiscalMonth = dt.Month - 6;
            }

            int FiscalQuarter = 0;
            if (FiscalMonth >= 1 && FiscalMonth <= 3)
            {
                FiscalQuarter = 1;
            }
            if (FiscalMonth >= 4 && FiscalMonth <= 6)
            {
                FiscalQuarter = 2;
            }
            if (FiscalMonth >= 7 && FiscalMonth <= 9)
            {
                FiscalQuarter = 3;
            }
            if (FiscalMonth >= 10 && FiscalMonth <= 12)
            {
                FiscalQuarter = 4;
            }

            return new FiscalPeriod(FiscalQuarter, FiscalMonth);
        }        [SqlUserDefinedType(typeof(FiscalPeriodFormatter))]
        public struct FiscalPeriod
        {
            public int Quarter { get; private set; }

            public int Month { get; private set; }

            public FiscalPeriod(int quarter, int month):this()
            {
                this.Quarter = quarter;
                this.Month = month;
            }

            public override bool Equals(object obj)
            {
                if (ReferenceEquals(null, obj))
                {
                    return false;
                }

                return obj is FiscalPeriod && Equals((FiscalPeriod)obj);
            }

            public bool Equals(FiscalPeriod other)
            {
return this.Quarter.Equals(other.Quarter) &&    this.Month.Equals(other.Month);
            }

            public bool GreaterThan(FiscalPeriod other)
            {
return this.Quarter.CompareTo(other.Quarter) > 0 || this.Month.CompareTo(other.Month) > 0;
            }

            public bool LessThan(FiscalPeriod other)
            {
return this.Quarter.CompareTo(other.Quarter) < 0 || this.Month.CompareTo(other.Month) < 0;
            }

            public override int GetHashCode()
            {
                unchecked
                {
                    return (this.Quarter.GetHashCode() * 397) ^ this.Month.GetHashCode();
                }
            }

            public static FiscalPeriod operator +(FiscalPeriod c1, FiscalPeriod c2)
            {
return new FiscalPeriod((c1.Quarter + c2.Quarter) > 4 ? (c1.Quarter + c2.Quarter)-4 : (c1.Quarter + c2.Quarter), (c1.Month + c2.Month) > 12 ? (c1.Month + c2.Month) - 12 : (c1.Month + c2.Month));
            }

            public static bool operator ==(FiscalPeriod c1, FiscalPeriod c2)
            {
                return c1.Equals(c2);
            }

            public static bool operator !=(FiscalPeriod c1, FiscalPeriod c2)
            {
                return !c1.Equals(c2);
            }
            public static bool operator >(FiscalPeriod c1, FiscalPeriod c2)
            {
                return c1.GreaterThan(c2);
            }
            public static bool operator <(FiscalPeriod c1, FiscalPeriod c2)
            {
                return c1.LessThan(c2);
            }
            public override string ToString()
            {
                return (String.Format("Q{0}:P{1}", this.Quarter, this.Month));
            }

        }

        public class FiscalPeriodFormatter : IFormatter<FiscalPeriod>
        {
public void Serialize(FiscalPeriod instance, IColumnWriter writer, ISerializationContext context)
            {
                using (var binaryWriter = new BinaryWriter(writer.BaseStream))
                {
                    binaryWriter.Write(instance.Quarter);
                    binaryWriter.Write(instance.Month);
                    binaryWriter.Flush();
                }
            }

public FiscalPeriod Deserialize(IColumnReader reader, ISerializationContext context)
            {
                using (var binaryReader = new BinaryReader(reader.BaseStream))
                {
var result = new FiscalPeriod(binaryReader.ReadInt16(), binaryReader.ReadInt16());
                    return result;
                }
            }
        }
    }
}

Używanie agregacji zdefiniowanych przez użytkownika: UDAGG

Agregacje zdefiniowane przez użytkownika to wszystkie funkcje związane z agregacją, które nie są dostarczane bez użycia języka U-SQL. Przykładem może być agregacja umożliwiająca wykonywanie niestandardowych obliczeń matematycznych, łączenie ciągów, manipulowanie ciągami itd.

Definicja zagregowanej klasy bazowej zdefiniowana przez użytkownika jest następująca:

    [SqlUserDefinedAggregate]
    public abstract class IAggregate<T1, T2, TResult> : IAggregate
    {
        protected IAggregate();

        public abstract void Accumulate(T1 t1, T2 t2);
        public abstract void Init();
        public abstract TResult Terminate();
    }

SqlUserDefinedAggregate wskazuje, że typ powinien być zarejestrowany jako agregacja zdefiniowana przez użytkownika. Tej klasy nie można dziedziczyć.

Atrybut SqlUserDefinedType jest opcjonalny dla definicji UDAGG.

Klasa bazowa pozwala przekazać trzy parametry abstrakcyjne: dwa jako parametry wejściowe i jeden jako wynik. Typy danych są zmienne i powinny być definiowane podczas dziedziczenia klas.

public class GuidAggregate : IAggregate<string, string, string>
{
    string guid_agg;

    public override void Init()
    { … }

    public override void Accumulate(string guid, string user)
    { … }

    public override string Terminate()
    { … }
}
  • Funkcja Init wywołuje raz dla każdej grupy podczas obliczeń. Zapewnia procedurę inicjowania dla każdej grupy agregacji.
  • Kumulacja jest wykonywana raz dla każdej wartości. Zapewnia ona główne funkcje algorytmu agregacji. Może służyć do agregowania wartości z różnymi typami danych zdefiniowanymi podczas dziedziczenia klas. Może akceptować dwa parametry zmiennych typów danych.
  • Zakończenie jest wykonywane raz na grupę agregacji na końcu przetwarzania w celu wyprowadzenia wyniku dla każdej grupy.

Aby zadeklarować poprawne typy danych wejściowych i wyjściowych, użyj definicji klasy w następujący sposób:

public abstract class IAggregate<T1, T2, TResult> : IAggregate
  • T1: Pierwszy parametr do zakumulowania
  • T2: Drugi parametr do zakumulowania
  • TResult: zwracany typ zakończenia

Na przykład:

public class GuidAggregate : IAggregate<string, int, int>

lub

public class GuidAggregate : IAggregate<string, string, string>

Używanie narzędzia UDAGG w języku U-SQL

Aby użyć narzędzia UDAGG, najpierw zdefiniuj go w kodzie lub odwołaj się do niego z istniejącej biblioteki DLL możliwości programowania, jak opisano wcześniej.

Następnie użyj następującej składni:

AGG<UDAGG_functionname>(param1,param2)

Oto przykład udaGG:

public class GuidAggregate : IAggregate<string, string, string>
{
    string guid_agg;

    public override void Init()
    {
        guid_agg = "";
    }

    public override void Accumulate(string guid, string user)
    {
        if (user.ToUpper()== "USER1")
        {
            guid_agg += "{" + guid + "}";
        }
    }

    public override string Terminate()
    {
        return guid_agg;
    }

}

I podstawowy skrypt U-SQL:

DECLARE @input_file string = @"\usql-programmability\input_file.tsv";
DECLARE @output_file string = @" \usql-programmability\output_file.tsv";

@rs0 =
    EXTRACT
            guid string,
            dt DateTime,
            user String,
            des String
    FROM @input_file
    USING Extractors.Tsv();

@rs1 =
    SELECT
        user,
        AGG<USQL_Programmability.GuidAggregate>(guid,user) AS guid_list
    FROM @rs0
    GROUP BY user;

OUTPUT @rs1 TO @output_file USING Outputters.Text();

W tym scenariuszu przypadków użycia łączymy identyfikatory GUID klasy dla określonych użytkowników.

Następne kroki