Tuesday, November 26, 2013

Handling Late Arriving/Open For Write Flat Files in SSIS

     If you have to source flat files for a data warehouse, then you may find yourself at the mercy of the source system for when these files are produced. If the times that these jobs complete vary, it can be quite a pain when trying to schedule your staging loads.  If you miss the window, or get a failure because the file is being written to when the job kicks off, then you have to start the job again. If there are data dependencies on these flat files in later parts of the job, then getting that data loaded into the data marts may have to be delayed as well. 

     What if you had a general idea of a window of time that these flat files were most likely to arrive. Say the flat files are supposed to be there every day at 5:00 pm, but never show up later than 5:30 pm. Wouldn't it be nice to have your load job be more flexible and give the file some time to show up before running the rest of the job, and not having to later bug a DBA to kick it off again to get the late arriving files? Well, luckily it's pretty easy to do, and I'll show you in this post how to do it.

     The control flow for this example is going to look like this:


Figure 1. Control Flow

    We're going to take advantage of a Script Task to do this. If we have success we process the file, if not we log it and/or shoot off an email. First thing we need to do is create our flat file connection manager for our test file. For this example I created a test directory in the root of c: called test. In the test directory I created a file called test.txt that the file connection can reference:


Figure 2. Flat File Connection Manager

    Next, we need to specify some configurations for our script task by using SSIS variables. The first variable we're going to create will be called minutesToCancel. This variable will tell the script task how many minutes to wait for the file to arrive in the directory before timing out. The second variable will be called secondsToTryAgain. This variable will tell the script task how many seconds to wait after not seeing the file in the directory before checking again to see if it arrived. In my example here I set the minutesToCancel to 1 minute and my secondsToTryAgain to 15 seconds. These can be changed to whatever you want based on your particular setup.


Figure 3. SSIS Variables
   
     With our flat file connection manager set up and SSIS variables created, we can start coding. Drag a script task onto the control flow design surface. On the Scipt screen, make sure to select our variables minutesToCancel and secondsToTryAgain as ReadOnlyVariables. This way the script task can access them in code. Click the Edit Script button so we can start coding.


Figure 4. Script Task Script Screen


Paste the following code into the script task:


#region Namespaces
using System;
using System.Data;
using Microsoft.SqlServer.Dts.Runtime;
using System.Windows.Forms;
using System.Threading.Tasks;
using System.Threading;
using System.IO;
#endregion

#region Class
/// <summary>
/// ScriptMain is the entry point class of the script.  Do not change the name, attributes,
/// or parent of this class.
/// </summary>
[Microsoft.SqlServer.Dts.Tasks.ScriptTask.SSISScriptTaskEntryPointAttribute]
public partial class ScriptMain : Microsoft.SqlServer.Dts.Tasks.ScriptTask.VSTARTScriptObjectModelBase
{

    //Connection manager for the flat file
    ConnectionManager cm;
    //The cancellation token for cancelling the task
    CancellationTokenSource tokenSource;
    //Boolean to determine if file was found
    bool connected = false;

    #region Methods
    /// <summary>
    /// This method is called when this script task executes in the control flow.
    /// Before returning from this method, set the value of Dts.TaskResult to indicate success or failure.
    /// To open Help, press F1.
    /// </summary>
    public void Main()
    {
        try
        {

            tokenSource = new CancellationTokenSource();
            const int Minutes = 60000;
            const int Seconds = 1000;
            //Set minutes before the package stops looking for the file
            int minutesToCancel = (int)Dts.Variables["minutesToCancel"].Value * Minutes;
            //Seconds for thread to wait before trying to see if the file arrived again
            int secondsToTryAgain = (int)Dts.Variables["secondsToTryAgain"].Value * Seconds;
            //Set connection manager to our flat file connection
            cm = Dts.Connections["Flat File Connection Manager"];
            //Set up method the task will call
            Action act = () => CheckforFile(tokenSource.Token, cm, connected, secondsToTryAgain);
            System.Threading.Tasks.Task task = new System.Threading.Tasks.Task(act);
            task.Start();
            //Timeout after specified minutes
            if (!task.Wait(minutesToCancel, tokenSource.Token))
            {
                Failure("Checking for file timedout");//can change to Success() with message if you dont want package to fail for timeout
            }
        }
        catch (OperationCanceledException)
        {
            //do nothing

        }

        catch (Exception e)
        {
            FailComponent(e.ToString());
        }

    }
    /// <summary>
    /// Loops until file arrives or timeout
    /// </summary>
    /// <param name="cancelToken">The token used to trigger task cancellation</param>
    /// <param name="cm">Connection manager for the flat file</param>
    /// <param name="connected">Boolean to set if file arrives</param>
    /// <param name="secondsToTryAgain">The number of milliseconds to wait before trying to see if the file arrived again</param>
    public void CheckforFile(CancellationToken cancelToken, ConnectionManager cm, bool connected, int secondsToTryAgain)
    {
        //Determine if file locked message needs to be resent
        bool resendMessage = true;

        //loop while we haven't triggered a cancellation and file hasn't arrived yet
        while (!cancelToken.IsCancellationRequested)
        {

            //If file arrives trigger success
            if (File.Exists(cm.ConnectionString))
            {
             
                try
                {
                    //Attempt to read first character of file
                    using (TextReader reader = File.OpenText(cm.ConnectionString))
                    {
                        char[] block = new char[1];
                        reader.ReadBlock(block, 0, 1);
                    }
                    //If file is available and not being written to mark success
                    connected = true;
                    Success("File found! Proceeding to process");
                }
                //If file is being written to catch exception
                catch (IOException)
                {
                    if (resendMessage)
                    {
                        InfoComponent("File arrived, but is still being written to");
                        resendMessage = false;
                    }
                }
            }

            //Sleep for specified seconds
            Thread.Sleep(secondsToTryAgain);
        }
    }

    /// <summary>
    /// Stops the while loop and reports success
    /// </summary>
    /// <param name="msg">Success message</param>
    public void Success(string msg)
    {
        if (!tokenSource.IsCancellationRequested)
        {
            InfoComponent(msg);
            tokenSource.Cancel();
            Dts.TaskResult = (int)ScriptResults.Success;

        }
    }

    /// <summary>
    /// Stops the while loop and reports failure
    /// </summary>
    /// <param name="msg">Failure Message</param>
    public void Failure(string msg)
    {
        if (!tokenSource.IsCancellationRequested)
        {
            FailComponent(msg);
            tokenSource.Cancel();
            Dts.TaskResult = (int)ScriptResults.Failure;

        }
    }

    /// <summary>Outputs an Error</summary>
    /// <param name="errorMsg">The error message to send to the UI</param>
    private void FailComponent(string errorMsg)
    {
        Dts.Events.FireError(0, "Error:", errorMsg, String.Empty, 0);
    }
    /// <summary>Outputs a Message</summary>
    /// <param name="errorMsg">The information message to send to the UI</param>
    private void InfoComponent(string infoMsg)
    {
        bool fail = false;
        Dts.Events.FireInformation(1, "Info:", infoMsg, "", 0, ref fail);
    }
    #endregion

    #region ScriptResults declaration
    /// <summary>
    /// This enum provides a convenient shorthand within the scope of this class for setting the
    /// result of the script.
    ///
    /// This code was generated automatically.
    /// </summary>
    enum ScriptResults
    {
        Success = Microsoft.SqlServer.Dts.Runtime.DTSExecResult.Success,
        Failure = Microsoft.SqlServer.Dts.Runtime.DTSExecResult.Failure
    };
    #endregion

}
#endregion


     Let's step through and explain some of this code. In order to keep a thread going and keep checking for the file to arrive we utilize a new feature that was introduced with the .net 4.0 library called Tasks. Tasks make anything to do with threads much easier than previous releases of .net. We declare a Task that will kick off a method that is going to keep checking for the arrival of the file in the directory. In addition to this we create a CancellationToken that is going to be passed to CheckForFile(). This token will let the while loop know when to break. The token is set to true through the CancellationTokenSource when the Cancel() method is called. While this token is set to false we keep looping, checking for a file, seeing if the file is still being written to, then putting the thread to sleep for 15 seconds. Once we time out, or the file arrives and is not being written to, we set the CancellationToken to true and exit the while loop.


     If the file is found, and not being written to, we output to the progress tab a file found message, then proceed to the data flow that will process this file:

Figure 5. Success Message
     If the file has not arrived before the time out threshold, we output an error message and fail the script task(if you don't want the script task to fail on time out call Success() instead of Failure() ):

Figure 6. Failure Message
     If the file has arrived, but is not done being written to, we continue to loop and we output that as well:


Figure 7. File Written to Message
     When the file is done being written to, we output success:


Figure 8. Success After File Finished Being Written

     This can save you a lot of headaches if you have jobs that miss the files arriving by a short time. It can also help you with SSIS packages that contain data flows with flat file sources that blow up when trying to read files that are still being written to from the source systems.

Saturday, November 23, 2013

Deserializing a JSON Feed with Nested Arrays in SSIS

     I thought I would expand upon the post I wrote about consuming JSON feeds in SSIS with a more complex piece of JSON. The JSON I used in the previous post was rather simple, but got to the general strategy of how to get SSIS to work with JSON. In this post I'd like to use JSON that have nested arrays and how we can deal with them in SSIS.  I found a site online that pushes out JSON feeds of research paper metadata at http://researchr.org/ . They have sample links such as  http://researchr.org/api/publication/HemelKGV-2010 which provides JSON that looks like this:

{
   
"abstract":"The realization of model-driven software development requires effective techniques for implementing code generators for domain-specific languages. This paper identifies techniques for improving separation of concerns in the implementation of generators. The core technique is code generation by model transformation, that is, the generation of a structured representation (model) of the target program instead of plain text. This approach enables the transformation of code after generation, which in turn enables the extension of the target language with features that allow better modularity in code generation rules. The technique can also be applied to \u2018internal code generation\u2019 for the translation of high-level extensions of a DSL to lower-level constructs within the same DSL using model-to-model transformations. This paper refines our earlier description of code generation by model transformation with an improved architecture for the composition of model-to-model normalization rules, solving the problem of combining type analysis and transformation. Instead of coarse-grained stages that alternate between normalization and type analysis, we have developed a new style of type analysis that can be integrated with normalizing transformations in a fine-grained manner. The normalization strategy has a simple extension interface and integrates non-local, context-sensitive transformation rules. We have applied the techniques in a realistic case study of domain-specific language engineering, i.e. the code generator for WebDSL, using Stratego, a high-level transformation language that integrates model-to-model, model-to-code, and code-to-code transformations. ",
   
"lastpage":402,
   
"type":"article",
   
"url":"http://researchr.org/publication/HemelKGV-2010",
   
"firstpage":375,
   
"issuenumber":"3",
   
"journal":"sosym",
   
"id":"c6ab7ca9-2764-4354-964b-ba8c3244bd88",
   
"authors":[
      
{
         
"person":{
            
"id":"947901fa-e910-454b-8546-c514c5a191be",
            
"fullname":"Zef  Hemel",
            
"key":"zefhemel",
            
"url":"http://researchr.org/profile/zefhemel"
         
},
         
"alias":{
            
"id":"7816809a-e2e6-42c6-8c44-481a821fb0bd",
            
"name":"Zef Hemel",
            
"key":"zef-hemel",
            
"url":"http://researchr.org/alias/zef-hemel"
         
}
      
},
      
{
         
"person":{
            
"id":"6e29bb96-28c9-4c76-915d-160ef1947602",
            
"fullname":"Lennart C. L. Kats",
            
"key":"lennartclkats",
            
"url":"http://researchr.org/profile/lennartclkats"
         
},
         
"alias":{
            
"id":"f6b6789b-a32a-417c-a90e-c37a12f25728",
            
"name":"Lennart C. L. Kats",
            
"key":"lennart-c.-l.-kats",
            
"url":"http://researchr.org/alias/lennart-c.-l.-kats"
         
}
      
},
      
{
         
"person":{
            
"id":"f50a3666-08fe-4b30-9734-181aed455bb1",
            
"fullname":"Danny M.  Groenewegen",
            
"key":"dannymgroenewegen",
            
"url":"http://researchr.org/profile/dannymgroenewegen"
         
},
         
"alias":{
            
"id":"46cf4818-8ce7-4644-9ee1-414e4c4960e7",
            
"name":"Danny M. Groenewegen",
            
"key":"danny-m.-groenewegen",
            
"url":"http://researchr.org/alias/danny-m.-groenewegen"
         
}
      
},
      
{
         
"person":{
            
"id":"f0fbf7c0-9729-4ec8-b3c9-5f30dbd9614b",
            
"fullname":"Eelco Visser",
            
"key":"eelcovisser",
            
"url":"http://researchr.org/profile/eelcovisser"
         
},
         
"alias":{
            
"id":"f68ba0ee-899e-4c4c-9d8a-6fed5092830a",
            
"name":"Eelco Visser",
            
"key":"eelco-visser",
            
"url":"http://researchr.org/alias/eelco-visser"
         
}
      
}
   
],
   
"title":"Code generation by model transformation: a case study in transformation modularity",
   
"month":"June",
   
"volumenumber":"9",
   
"year":"2010",
   
"note":"",
   
"key":"HemelKGV-2010",
   
"doi":"http://dx.doi.org/10.1007/s10270-009-0136-1"
}

     Here we can see the attributes of the paper like the abstract, the url of the paper, the date it was published, etc.  But what we also see is a nested array of the authors of the paper. Each author has a nested person object and alias object. So we have a fairly complex piece of JSON to work with, but not so complex that I can't clearly describe how to deal with it. 

      Our data flow for this SSIS package is going to look something like this:


Figure 1. Data Flow

We have outputs for attributes of the main paper, the person attributes of the authors as well as the attributes of the alias of the authors. So, let's drag a script component onto the design surface and begin building this solution. When prompted to select a type, pick source:


Figure 2. Script Component Type
Click on Inputs and Outputs to pull up this screen, this is where we will configure our outputs. First we build out the default output for the attributes of the paper. All the fields are strings (DT_STR) except for abstractnote and note which i made DT_TEXT, as well as lastpage and firstpage which I made integers(DT_I4):


Figure 3. Script Component Inputs and Outputs Screen(Paper)
     The output for both Author Person and Author Alias will match what is in the JSON except for 2 fields. I added parentpaperid to Author Person so we could link the author data back to the paper and I added personid to the Author Alias so I could link back the alias to the person:


Figure 4. Script Component Inputs and Outputs Screen (Authors)

From reading my previous post on this, you'll know that we will need to create the classes that this JSON feed will need to deserialize into. Luckily, I was recently shown a site that makes this very easy. When copying the JSON text to json2csharp.com,  it  produced for me these classes to derserialize my JSON into:


public class Person
{
   public string id { get; set; }
    public string fullname { get; set; }
    public string key { get; set; }
    public string url { get; set; }
}

public class Alias
{
    public string id { get; set; }
    public string name { get; set; }
    public string key { get; set; }
    public string url { get; set; }
}

public class Author
{
    public Person person { get; set; }
    public Alias alias { get; set; }
}

public class RootObject
{
    public string @abstract { get; set; }
    public int lastpage { get; set; }
    public string type { get; set; }
    public string url { get; set; }
    public int firstpage { get; set; }
    public string issuenumber { get; set; }
    public string journal { get; set; }
    public string id { get; set; }
    public List<Author> authors { get; set; }
    public string title { get; set; }
    public string month { get; set; }
    public string volumenumber { get; set; }
    public string year { get; set; }
    public string note { get; set; }
    public string key { get; set; }
    public string doi { get; set; }
}
     These classes will hold all of our data and allow it to be outputted to our 3 output buffers we created. So now go back to the script screen, click on the Edit Script button, and we can start some coding (be aware that you need to add System.Web.Extensions.dll as a reference, I wrote a step by step on how to do this in my last post regarding JSON here):

#region Namespaces
using System;
using System.Data;
using Microsoft.SqlServer.Dts.Pipeline.Wrapper;
using Microsoft.SqlServer.Dts.Runtime.Wrapper;
using System.Net;
using Microsoft.SqlServer.Dts.Runtime;
using System.Windows.Forms;
using System.IO;
using System.Web.Script.Serialization;
using System.Collections.Generic;
using System.Text;
#endregion

#region Class
[Microsoft.SqlServer.Dts.Pipeline.SSISScriptComponentEntryPointAttribute]
public class ScriptMain : UserComponent
{

    #region Methods
    /// <summary>Outputs records to the output buffer</summary>
    public override void CreateNewOutputRows()
    {

        //Set Webservice URL
        string wUrl = "http://researchr.org/api/publication/HemelKGV-2010";

        try
        {
            //Call getWebServiceResult to return our paper
            RootObject outPutResponse = GetWebServiceResult(wUrl);
            //Output main attributes of paper
            Output0Buffer.AddRow();
            if (outPutResponse.@abstract == null)
            {
                Output0Buffer.abstractnote.SetNull();
            }
            else
            {
                Output0Buffer.abstractnote.AddBlobData(Encoding.ASCII.GetBytes(outPutResponse.@abstract));
            }
            Output0Buffer.lastpage = outPutResponse.lastpage;
            Output0Buffer.type = outPutResponse.type;
            Output0Buffer.url = outPutResponse.url;
            Output0Buffer.firstpage = outPutResponse.firstpage;
            Output0Buffer.issuenumber = outPutResponse.issuenumber;
            Output0Buffer.journal = outPutResponse.journal;
            Output0Buffer.id = outPutResponse.id;
            Output0Buffer.title = outPutResponse.title;
            Output0Buffer.month = outPutResponse.month;
            Output0Buffer.volumenumber = outPutResponse.volumenumber;
            Output0Buffer.year = outPutResponse.year;
            if (outPutResponse.note == null)
            {
                Output0Buffer.note.SetNull();
            }
            else
            {
                Output0Buffer.note.AddBlobData(Encoding.ASCII.GetBytes(outPutResponse.note));
            }
            Output0Buffer.key = outPutResponse.key;
            Output0Buffer.doi = outPutResponse.doi;
            //Output author info
            foreach (Author auth in outPutResponse.authors)
            {
                //Output person info
                AuthorPersonBuffer.AddRow();
                AuthorPersonBuffer.id = auth.person.id;
                AuthorPersonBuffer.fullname = auth.person.fullname;
                AuthorPersonBuffer.key = auth.person.key;
                AuthorPersonBuffer.url = auth.person.url;
                AuthorPersonBuffer.parentpaperid = outPutResponse.id;//link person to main paper

                //Output alias info
                AuthorAliasBuffer.AddRow();
                AuthorAliasBuffer.id = auth.alias.id;
                AuthorAliasBuffer.name = auth.alias.name;
                AuthorAliasBuffer.key = auth.alias.key;
                AuthorAliasBuffer.url = auth.alias.url;
                AuthorAliasBuffer.personid = auth.person.id;//link alias to person
            }

        }
        catch (Exception e)
        {
            FailComponent(e.ToString());
        }

    }

    /// <summary>
    /// Method to return our research paper
    /// </summary>
    /// <param name="wUrl">The web service URL to call</param>
    /// <returns>A single research paper</returns>
    private RootObject GetWebServiceResult(string wUrl)
    {

        HttpWebRequest httpWReq = (HttpWebRequest)WebRequest.Create(wUrl);
        HttpWebResponse httpWResp = (HttpWebResponse)httpWReq.GetResponse();
        RootObject jsonResponse = null;

        try
        {
            //Test the connection
            if (httpWResp.StatusCode == HttpStatusCode.OK)
            {

                Stream responseStream = httpWResp.GetResponseStream();
                string jsonString = null;

                //Set jsonString using a stream reader
                using (StreamReader reader = new StreamReader(responseStream))
                {
                    jsonString = reader.ReadToEnd();
                    reader.Close();
                }

                //Deserialize our JSON
                JavaScriptSerializer sr = new JavaScriptSerializer();
                jsonResponse = sr.Deserialize<RootObject>(jsonString);

            }
            //Output connection error message
            else
            {
                FailComponent(httpWResp.StatusCode.ToString());

            }
        }
        //Output JSON parsing error
        catch (Exception e)
        {
            FailComponent(e.ToString());
        }
        return jsonResponse;

    }

    /// <summary>
    /// Outputs error message
    /// </summary>
    /// <param name="errorMsg">Full error text</param>
    private void FailComponent(string errorMsg)
    {
        bool fail = false;
        IDTSComponentMetaData100 compMetadata = this.ComponentMetaData;
        compMetadata.FireError(1, "Error Getting Data From Webservice!", errorMsg, "", 0, out fail);

    }
    #endregion
}
#endregion

#region JSON Classes
//Attributes of the author
public class Person
{

    public string id { getset; }

    public string fullname { getset; }

    public string key { getset; }

    public string url { getset; }

}
//Alias of the author
public class Alias
{

    public string id { getset; }

    public string name { getset; }

    public string key { getset; }

    public string url { getset; }

}
//Authors of the paper
public class Author
{

    public Person person { getset; }

    public Alias alias { getset; }

}
//Root for Paper
public class RootObject
{

    public string @abstract { getset; }

    public int lastpage { getset; }

    public string type { getset; }

    public string url { getset; }

    public int firstpage { getset; }

    public string issuenumber { getset; }

    public string journal { getset; }

    public string id { getset; }

    public List<Author> authors { getset; }

    public string title { getset; }

    public string month { getset; }

    public string volumenumber { getset; }

    public string year { getset; }

    public string note { getset; }

    public string key { getset; }

    public string doi { getset; }

}

#endregion

     Let's step through the output and explain what's going on here. After we deserialize our JSON into our various classes jsonResponse = sr.Deserialize<RootObject>(jsonString); (keep in mind we have one paper coming back, if we had multiple you would want to change the generic to an array or list) we then output our attributes for the paper to the Ouput0Buffer:
//Call getWebServiceResult to return our paper
RootObject outPutResponse = GetWebServiceResult(wUrl);
//Output main attributes of paper
  Output0Buffer.AddRow();
  if (outPutResponse.@abstract == null)
  {
     Output0Buffer.abstractnote.SetNull();
  }
  else
  {
   Output0Buffer.abstractnote.AddBlobData(Encoding.ASCII.GetBytes(outPutResponse.@abstract));
  }
  Output0Buffer.lastpage = outPutResponse.lastpage;
 .....

After that we need to loop through the Authors list object and output our person and alias attributes, starting with person:

 foreach (Author auth in outPutResponse.authors)
            {
                //Output person info
                AuthorPersonBuffer.AddRow();
                AuthorPersonBuffer.id = auth.person.id;
                AuthorPersonBuffer.fullname = auth.person.fullname;
                AuthorPersonBuffer.key = auth.person.key;
                AuthorPersonBuffer.url = auth.person.url;
                AuthorPersonBuffer.parentpaperid = outPutResponse.id; 

These match the attributes of Person except for one attribute that I added  AuthorPersonBuffer.parentpaperid = outPutResponse.id; which is the id attribute of the paper so I can link this author to the paper. Next, we output our alias data:

                //Output alias info
                AuthorAliasBuffer.AddRow();
                AuthorAliasBuffer.id = auth.alias.id;
                AuthorAliasBuffer.name = auth.alias.name;
                AuthorAliasBuffer.key = auth.alias.key;
                AuthorAliasBuffer.url = auth.alias.url;
                AuthorAliasBuffer.personid = auth.person.id;

These match the attributes of Alias except for one attribute that I added  AuthorAliasBuffer.personid = auth.person.id; which is the id attribute of the person I can link this alias back to.

     With coding completed let's save, build and run the package. (I added data viewers to all of the outputs so we can see the results):

Figure 5. Data Viewers

     Here you can see all of the data we pulled out of that JSON in our data flow. When arriving at database table destinations you can use SQL to join the main paper to author persons on id = parentpaperid and author persons to author alias on id = personid(This is definitely not the best database design, but suits the example as a visual and is only intended for staging. You would want to apply some transformations on this data once its at the destination to tidy up the design) :


Figure 6. Destination Tables