Join: Example of Creating

This article describes an example of creating and executing an ETL task with joining of data from two sources.

Three tables must be in the repository: T_SourceOne, T_SourceTwo and T_Destination. Structure of sources must be identical and must contain the ID and VALUE fields. The structure of data consumer must contain the ID, VALUE1 and VALUE2 fields. The repository must also contain an ETL task with the ETLTASKS identifier. On executing the example presented below four objects are created in the ETL task: two repository providers, the Join transformer and a repository consumer. The required properties and links are set for all objects:

After objects are created and saved, the ETL task is executed. The similar code applied to different objects is placed into separate procedures or functions.

Add links to the Andy, Db, Drawing, Dt, Etl, Metabase system assemblies.

Sub UserProc;
Var
    MB: IMetabase;
    ETLTask: IEtlTask;
    EtlProviderOne, EtlProviderTwo: IEtlPlainDataProvider;
    MetabaseProviderOne, MetabaseProviderTwo: IDtMetabaseProvider;
    EtlConsumer: IEtlPlainDataConsumer;
    MetabaseConsumer: IDtMetabaseConsumer;
    Join: IEtlPlainDataJoin;
    JoinInputs: IEtlPlainInputs;
    InputOne, InputTwo: IEtlPlainInput;
    Link, Link1, Link2: IEtlPlainLink;
    Shapes: IWxShapes;

    ProvOneFields, ProvTwoFields, JoinInOneFields, JoinInTwoFields, JoinOutFields, ConsFields: IEtlPlainFields;
    Index: IEtlPlainIndex;
    Mapper: IEtlPlainFieldsMapper;
Begin
//Open ETL task
    MB := MetabaseClass.Active;
    ETLTask := MB.ItemById("ETLTASKS").Edit As IEtlTask;
//Create the first provider
    EtlProviderOne := ETLTask.Create(EtlObjectType.PlainDataMetabaseProvider) As IEtlPlainDataProvider;
    EtlProviderOne := EtlProviderOne.Edit;
    EtlProviderOne.Id := "Metabase_Provider1";
    EtlProviderOne.Name := "Import from table 1";

    MetabaseProviderOne := EtlProviderOne.Provider As IDtMetabaseProvider;
    MetabaseProviderOne.Dataset := MB.ItemById("T_SourceOne").Bind As IDatasetModel;
    EtlProviderOne.FillDefault;
     //Create a visual object of provider
    CreateWX(ETLTask, EtlProviderOne, -60, 0);
//Create the second provider
    EtlProviderTwo := ETLTask.Create(EtlObjectType.PlainDataMetabaseProvider) As IEtlPlainDataProvider;
    EtlProviderTwo := EtlProviderTwo.Edit;
    EtlProviderTwo.Id := "Metabase_Provider2";
    EtlProviderTwo.Name := "Import from table 2";
    MetabaseProviderTwo := EtlProviderTwo.Provider As IDtMetabaseProvider;
    MetabaseProviderTwo.Dataset := MB.ItemById("T_SourceTwo").Bind As IDatasetModel;
    EtlProviderTwo.FillDefault;

     //Create a visual object of data provider
    CreateWX(ETLTask, EtlProviderTwo, -60, 20);
//Create a data consumer
    EtlConsumer := ETLTask.Create(EtlObjectType.PlainDataMetabaseConsumer) As IEtlPlainDataConsumer;
    EtlConsumer := EtlConsumer.Edit;
    EtlConsumer.Id := "Metabase_Consumer";
    EtlConsumer.Name := "Export to table";
    MetabaseConsumer := EtlConsumer.Consumer As IDtMetabaseConsumer;
    MetabaseConsumer.Dataset := MB.ItemById("T_Destination").Bind As IDatasetModel;
    EtlConsumer.FillDefault;

     //Create a visual object of data consumer
    CreateWX(ETLTask, EtlConsumer, 60, 0);
//Create the Join transformer
    Join := ETLTask.Create(EtlObjectType.PlainDataJoin) As IEtlPlainDataJoin;
    Join := Join.Edit;
    Join.Id := "Join_Transform";
    Join.Name := "Join";
     //Create two outputs
    JoinInputs := Join.PlainInputs;
    JoinInputs.Add;
    JoinInputs.Add;
    InputOne := JoinInputs.Item(0);
    InputTwo := JoinInputs.Item(1);
     //Create a visual object of data transformer
    CreateWX(ETLTask, Join, 0, 0);
//Fill in lists of fields of all objects based on list of data provider fields
     //Get list of fields
    ProvOneFields := EtlProviderOne.PlainOutput.Fields; //First data provider output
    ProvTwoFields := EtlProviderTwo.PlainOutput.Fields; //Second data provider output
    JoinInOneFields := InputOne.Fields; //First data transformer input
    JoinInTwoFields := InputTwo.Fields; //Second data transformer input

    JoinOutFields := Join.PlainOutput.Fields; //Transformer output
    ConsFields := EtlConsumer.PlainInput.Fields; //Consumer input
     //Fill lists of fields
    FillFields(JoinInOneFields, ProvOneFields);
    FillFields(JoinInTwoFields, ProvTwoFields);
    FillFields(JoinOutFields, ConsFields);
     //Set up conditions and link fields
    Join.Type := EtlJoinType.Outer;
    Join.PrimaryInput := InputOne;
    Index := Join.Index.Edit;
    Index.AddField;
    Index.PlainInputField(0, 0) := InputOne.Fields.FindById("ID");
    Index.PlainInputField(0, 1) := InputTwo.Fields.FindById("ID");
    Index.Save;
    Mapper := Join.Mapper;
    Mapper.Map(JoinOutFields.FindById("ID")).AsString := InputOne.Id + ".ID";
    Mapper.Map(JoinOutFields.FindById("VALUE1")).AsString := InputOne.Id + ".VALUE";
    Mapper.Map(JoinOutFields.FindById("VALUE2")).AsString := InputTwo.Id + ".VALUE";

//Save all objects
    EtlProviderOne.Save;
    EtlProviderTwo.Save;
    Join.Save;
    EtlConsumer.Save;
//Link of transformer with all objects
     //Create links
    Link := CreateLink(ETLTask, EtlProviderOne.PlainOutput, InputOne);
    Link1 := CreateLink(ETLTask, EtlProviderTwo.PlainOutput, InputTwo);
    Link2 := CreateLink(ETLTask, Join.PlainOutput, EtlConsumer.PlainInput);
     //Create visual links
    Shapes := ETLTask.Workspace.Shapes;
    CreateWXLink(ETLTask, Shapes.FindById(EtlProviderOne.Id), Shapes.FindById(Join.Id), Link);

    CreateWXLink(ETLTask, Shapes.FindById(EtlProviderTwo.Id), Shapes.FindById(Join.Id), Link1);
    CreateWXLink(ETLTask, Shapes.FindById(Join.Id), Shapes.FindById(EtlConsumer.Id), Link2);
//Save task
    (ETLTask As IMetabaseObject).Save;
//Execute task
    ETLTask.Execute(Null);
End Sub UserProc;
//Create visual objects
Sub CreateWX(ETLTask: IEtlTask; ETLObject: IEtlObject; XPosition: Integer; YPosition: Integer);
Var
    WxRect: IWxRectangle;
    WxETLObj: IWxEtlObject;
Begin
    WxRect := ETLTask.Workspace.CreateRectangle;

    WxRect.Id := ETLObject.Id;
    WxETLObj := New WxEtlObject.Create;
    WxETLObj.EtlObject := ETLObject;
    WxRect.Style.TextPosition := WxTextPosition.Bottom;
    WxRect.Style.PictureMarginTop := -10;
    WxRect.PinPosition := New GxPointF.Create(XPosition, YPosition);
    WxRect.Extension := WxETLObj As IWxShapeExtension;
End Sub CreateWX;
//Fill lists of input/output fields
Sub FillFields(Fields, FieldsSource: IEtlPlainFields);
Begin
    Fields := Fields.Edit;
    Fields.Fill(FieldsSource);
    Fields.Save;
End Sub FillFields;
//Create a link between objects
Function CreateLink(ETLTask: IEtlTask; Output: IEtlPlainOutput; Input: IEtlPlainInput): IEtlPlainLink;
Var
    Link: IEtlPlainLink;

Begin
    Link := ETLTask.CreatePlainLink;
    Link.SourceObjectOutput := Output;
    Link.DestinationObjectInput := Input;
    Link.FillDefault;
    Return Link;
End Function CreateLink;
//Create visual links
Sub CreateWXLink(ETLTask: IEtlTask; Shape1: IWxShape; Shape2: IWxShape; Link: IEtlPlainLink);
Var
    WLink: IWxLink;
    WxETLLink: IWxEtlObject;
Begin

     //Create visual link objects
    WLink := ETLTask.Workspace.AutoLinkShapes(Shape1, Shape2);
    WLink.Style.LinePenBeginWxCap := WxLineCap.Flat;
    WLink.Style.LinePenEndWxCap := WxLineCap.Arrow30DegreeFilled;
    WxETLLink := New WxEtlObject.Create;
    WxETLLink.EtlObject := Link;
    WLink.Extension := WxETLLink As IWxShapeExtension;
End Sub CreateWXLink;