U-SQL programozhatósági útmutató – UDT és UDAGG

  • Cikk

Felhasználó által definiált típusok használata: UDT

A felhasználó által definiált típusok vagy UDT az U-SQL másik programozhatósági funkciója. Az U-SQL UDT normál C# felhasználó által definiált típusként működik. A C# egy erősen gépelt nyelv, amely lehetővé teszi a beépített és egyéni felhasználó által definiált típusok használatát.

Az U-SQL nem tudja implicit módon szerializálni vagy deszerializálni az tetszőleges UDT-ket, ha az UDT a sorhalmazok csúcsai között van átadva. Ez azt jelenti, hogy a felhasználónak explicit formázót kell megadnia az IFormatter felület használatával. Ez biztosítja az U-SQL-t az UDT szerializálási és deszerializálási metódusaival.

Megjegyzés

Az U-SQL beépített kinyerői és kimeneti elemei jelenleg nem szerializálják vagy deszerializálják az UDT-adatokat a fájlokba vagy fájlokból még az IFormatter-készlettel sem. Ha tehát UDT-adatokat ír egy OUTPUT utasítással rendelkező fájlba, vagy egy kiválasztóval olvassa be, sztringként vagy bájttömbként kell átadnia. Ezután explicit módon meghívja a szerializálási és deszerializálási kódot (azaz az UDT ToString() metódusát). A felhasználó által definiált kiválasztók és kimenetik viszont képesek az UDT-k olvasására és írására.

Ha UDT-t próbálunk használni a EXTRACTOR vagy a OUTPUTTER alkalmazásban (a korábbi SELECT-ből), az itt látható módon:

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

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

A következő hibaüzenet jelenik meg:

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

Az UDT kimenetben való használatához vagy sztringre kell szerializálnunk a ToString() metódussal, vagy létre kell hoznunk egy egyéni kimenetit.

Az UDT-k jelenleg nem használhatók a GROUP BY-ban. Ha UDT-t használ a GROUP BY-ban, a következő hiba jelenik meg:

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

Az UDT meghatározásához a következőket kell tennie:

  1. Adja hozzá a következő névtereket:
using Microsoft.Analytics.Interfaces
using System.IO;

  1. Adja hozzá Microsoft.Analytics.Interfacesaz UDT-felületekhez szükséges elemet. System.IO Emellett szükség lehet az IFormatter-felület meghatározására is.

  2. Definiáljon egy használt típust az SqlUserDefinedType attribútummal.

Az SqlUserDefinedType használatával a szerelvényben lévő típusdefiníciók felhasználói típusként (UDT) jelölhetők meg az U-SQL-ben. Az attribútum tulajdonságai az UDT fizikai jellemzőit tükrözik. Ez az osztály nem örökölhető.

Az SqlUserDefinedType az UDT-definíció kötelező attribútuma.

Az osztály konstruktora:

  • SqlUserDefinedTypeAttribute (type formatter)

  • Típusformázás: Az UDT-formázó definiálásához szükséges paraméter – pontosabban az IFormatter illesztő típusát itt kell megadni.

[SqlUserDefinedType(typeof(MyTypeFormatter))]
public class MyType
{ … }
  • A tipikus UDT-hez szükség van az IFormatter felület definíciójára is, ahogy az a következő példában is látható:
public class MyTypeFormatter : IFormatter<MyType>
{
    public void Serialize(MyType instance, IColumnWriter writer, ISerializationContext context)
    { … }

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

Az IFormatter interfész szerializál és deszerializál egy objektumgráfot a typeparamref name="T"> gyökértípussal<.

<typeparam name="T">A szerializálni és deszerializálni kívánt objektumgráf gyökértípusa.

  • Deszerializálás: Szerializálja az adatokat a megadott adatfolyamon, és újrakonfigurálja az objektumok gráfját.

  • Szerializálás: Szerializál egy objektumot vagy objektumgráfot a megadott adatfolyam adott gyökerével.

MyType instance: A típus példánya. IColumnWriter író/ IColumnReader olvasó: Az alapul szolgáló oszlopstream. ISerializationContext context: Enum, amely a stream forrás- vagy célkörnyezetét a szerializálás során meghatározó jelzőkészletet definiálja.

  • Köztes: Azt adja meg, hogy a forrás- vagy célkörnyezet nem állandó tároló.

  • Adatmegőrzés: Azt adja meg, hogy a forrás- vagy célkörnyezet egy megőrzött tároló.

Normál C#-típusként az U-SQL UDT-definíciók tartalmazhatnak felülbírálásokat az olyan operátorokhoz, mint a +/==/!=. Statikus metódusokat is tartalmazhat. Ha például ezt az UDT-t egy U-SQL MIN összesítő függvény paramétereként fogjuk használni, operátor-felülbírálást kell definiálnunk < .

Az útmutató korábbi részében bemutattunk egy példát a pénzügyi időszaknak az adott dátumból való azonosítására a formátumban Qn:Pn (Q1:P10). Az alábbi példa bemutatja, hogyan definiálhat egyéni típust a pénzügyi időszak értékeihez.

Az alábbiakban egy példa látható egy kód mögötti szakaszra, amely egyéni UDT- és IFormatter-felülettel rendelkezik:

[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;
        }
    }
}

A definiált típus két számot tartalmaz: negyedév és hónap. Az operátorok ==/!=/>/< és a statikus metódus ToString() itt van definiálva.

Ahogy korábban említettük, az UDT használható a SELECT kifejezésekben, de egyéni szerializálás nélkül nem használható a OUTPUTTER/EXTRACTOR függvényben. Vagy sztringként ToString() kell szerializálni egy egyéni OUTPUTTER/EXTRACTOR használatával.

Most nézzük meg az UDT használatát. Egy kód mögötti szakaszban a GetFiscalPeriod függvényt a következőre módosítottuk:

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);
}

Amint látható, a FiscalPeriod típus értékét adja vissza.

Itt bemutatunk egy példát arra, hogyan használhatja tovább az U-SQL alapszkriptjében. Ez a példa az U-SQL-szkripttől eltérő UDT-hívásokat mutat be.

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();

Íme egy példa egy teljes kód mögötti szakaszra:

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;
                }
            }
        }
    }
}

Felhasználó által definiált összesítések használata: UDAGG

A felhasználó által definiált összesítések olyan aggregációval kapcsolatos függvények, amelyek nem kerülnek házon kívülre az U-SQL-sel. A példa lehet egy aggregátum, amely egyéni matematikai számításokat, sztringösszefűzéseket, sztringekkel végzett manipulációkat stb. hajt végre.

A felhasználó által definiált összesítő alaposztály definíciója a következő:

    [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();
    }

Az SqlUserDefinedAggregate azt jelzi, hogy a típust felhasználó által definiált összesítésként kell regisztrálni. Ez az osztály nem örökölhető.

Az SqlUserDefinedType attribútum nem kötelező az UDAGG-definícióhoz.

Az alaposztály három absztrakt paraméter átadását teszi lehetővé: kettőt bemeneti paraméterekként, egyet pedig eredményként. Az adattípusok változók, és az osztályöröklés során kell definiálni.

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()
    { … }
}
  • Az Init a számítás során minden csoporthoz egyszer hív meg meghívást. Inicializálási rutint biztosít az egyes összesítési csoportokhoz.
  • A halmozódás minden értékhez egyszer lesz végrehajtva. Ez biztosítja az összesítő algoritmus fő funkcióit. Az osztályöröklés során definiált különböző adattípusokkal rendelkező értékek összesítésére használható. A változó adattípusok két paraméterét is elfogadhatja.
  • A megszakítást a rendszer a feldolgozás végén összesítő csoportonként egyszer hajtja végre az egyes csoportok eredményének kimenete érdekében.

A helyes bemeneti és kimeneti adattípusok deklarálásához használja az osztálydefiníciót az alábbiak szerint:

public abstract class IAggregate<T1, T2, TResult> : IAggregate
  • T1: Az első gyűjtendő paraméter
  • T2: A második halmozandó paraméter
  • TResult: A megszakítás visszatérési típusa

Például:

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

vagy

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

Az UDAGG használata az U-SQL-ben

Az UDAGG használatához először definiálja a kód mögötti kódban, vagy hivatkozzon rá a már létező programozhatósági DLL-ről a korábban ismertetett módon.

Ezután használja a következő szintaxist:

AGG<UDAGG_functionname>(param1,param2)

Íme egy példa az UDAGG-ra:

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;
    }

}

És alap U-SQL-szkript:

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();

Ebben a használati esetben összefűzzük az osztály GUID azonosítóit az adott felhasználók számára.

Következő lépések