Loading JSON Files with Nested Arrays from Azure Blob Storage into Hive Tables in HDInsight

     In my previous post I wrote about how to upload JSON files into Azure blob storage. In this post, I'd like to expand upon that and show how to load these files into Hive in Azure HDInsight.

     
     The 2 JSON files I'm going to load are up in my blob storage, acronym/abc.txt and acronym/def.txt:

Figure 1. JSON Files in Azure Blob Storage

 Both of these files came from a public RESTful API:

http://www.nactem.ac.uk/software/acromine/dictionary.py?sf=abc

and

http://www.nactem.ac.uk/software/acromine/dictionary.py?sf=def


The format for these 2 files looks something like this:

{  
   "a":[  
      {  
         "sf":"ABC",
         "lfs":[  
            {  
               "lf":"ATP-binding cassette",
               "freq":1437,
               "since":1990,
               "vars":[  
                  {  
                     "lf":"ATP-binding cassette",
                     "freq":1057,
                     "since":1990
                  }
               ]
            }
         ]
      }
   ]
}

     First thing I'd like to do is create an external table in Hive, where I'm going to "load" the raw JSON files, so we can play around a little with some of the out of box Hive functions for JSON. In the Hive Query Editor 

DROP TABLE IF EXISTS AcronymRaw;

CREATE EXTERNAL TABLE AcronymRaw(json string)

STORED AS TEXTFILE LOCATION 'wasb://jymbo@jymbo.blob.core.windows.net/acronym/';

If we execute the following we can see the raw JSON:

SET hive.execution.engine=tez;

SELECT * FROM  AcronymRaw;

We will get back the raw JSON from the files. (I have to add the tez switch since I did not configure my cluster to use this engine by default). If I want to dive into the first array of my JSON objects and see the acronyms I can use a lateral view, to flatten out the hierarchy, combined with a  json_tuple function:

SET hive.execution.engine=tez;

SELECT

  b.sf

 FROM AcronymRaw AS a lateral view json_tuple(get_json_object(a.json, '$.a[0]'),'sf') b AS sf

Will give me:

Figure 2. JSON Acronym Results

If we want to step into the first array of the next level in the hierarchy, lfs, we can add another lateral view to the statement:

SET hive.execution.engine=tez;

SET hive.cli.print.header=true;

SELECT

  b.sf,

  c.lf,

  c.freq,

  c.since

 FROM AcronymRaw as a lateral view json_tuple(get_json_object(a.json, '$.a[0]'), 'sf', 'lfs') b AS sf, lfs

lateral view json_tuple(get_json_object(concat('{"a":', b.lfs, '}'), '$.a[0]'), 'lf', 'freq', 'since') c AS lf, freq, since;

Figure 3. First Nested JSON Array



     Now to get this JSON loaded into an external table, that will match the structure of the JSON, we're going to incorporate a 3rd party java library that can deserialize the JSON for us. This will make querying the data much easier. You can download and build the project using Maven from this github link, or you can be lazy, like me, and get the JARS from here. Using Visual Studio you can upload these files to blob storage so they can be referenced in Hive:

Figure 4. Visual Studio Server Explorer

       
     After adding these files to Blob storage we can reference them in our Hive statements when creating and referencing tables using this library:

ADD JAR wasb://jymbo@jymbo.blob.core.windows.net/user/hdp/share/lib/hive/json-serde-1.1.9.3-SNAPSHOT-jar-with-dependencies.jar;

ADD JAR wasb://jymbo@jymbo.blob.core.windows.net/user/hdp/share/lib/hive/json-serde-1.1.9.3-SNAPSHOT.jar;

DROP TABLE IF EXISTS AcronymNonNormalized;

CREATE EXTERNAL TABLE AcronymNonNormalized (

  a array<

    struct<sf:string,lfs:array<

          struct<lf:string, freq:int, since:int, vars:array<

                struct<lf:string, freq:int, since:int>

                  >

                >

              >

            >

          >

)

ROW FORMAT SERDE 'org.openx.data.jsonserde.JsonSerDe'

STORED AS TEXTFILE LOCATION 'wasb://jymbo@jymbo.blob.core.windows.net/acronym/';

     Now that this structure is in a Hive table, we can query it much easier than in the raw JSON format. In this query we can explode the lfs arrays and project them alongside sf:

ADD JAR wasb://jymbo@jymbo.blob.core.windows.net/user/hdp/share/lib/hive/json-serde-1.1.9.3-SNAPSHOT-jar-with-dependencies.jar;

ADD JAR wasb://jymbo@jymbo.blob.core.windows.net/user/hdp/share/lib/hive/json-serde-1.1.9.3-SNAPSHOT.jar;

SET hive.execution.engine=tez;

SET hive.cli.print.header=true;

SELECT

 

    a[0].sf,

    lfs.lf,

    lfs.freq,

    lfs.since

 

FROM AcronymNonNormalized LATERAL VIEW EXPLODE( a[0].lfs) lfstable AS lfs;

Figure 5. Exploded Results

     We can even drill into the second tier array vars using another lateral view and explode:

ADD JAR wasb://jymbo@jymbo.blob.core.windows.net/user/hdp/share/lib/hive/json-serde-1.1.9.3-SNAPSHOT-jar-with-dependencies.jar;

ADD JAR wasb://jymbo@jymbo.blob.core.windows.net/user/hdp/share/lib/hive/json-serde-1.1.9.3-SNAPSHOT.jar;

SET hive.execution.engine=tez;

SET hive.cli.print.header=true;

SELECT

 

    a[0].sf,

    lfs.lf as mainlf,

    vars.lf,

    vars.freq,

    vars.since

 

  FROM AcronymNonNormalized LATERAL VIEW EXPLODE( a[0].lfs) lfstable AS lfs

  LATERAL VIEW EXPLODE(lfstable.lfs.vars) vartable AS vars;

Figure 6. Second Nested Array

     
     So now that we've figured out how to traverse the arrays in the table, we could potentially normalize this table and create a schema similar to this:

Figure 7. Normalized Schema

 The first table we need to populate will be AcronymSF:

   ADD JAR wasb://jymbo@jymbo.blob.core.windows.net/user/hdp/share/lib/hive/json-serde-1.1.9.3-SNAPSHOT-jar-with-dependencies.jar;

ADD JAR wasb://jymbo@jymbo.blob.core.windows.net/user/hdp/share/lib/hive/json-serde-1.1.9.3-SNAPSHOT.jar;

SET hive.execution.engine=tez;

SET hive.cli.print.header=true;

DROP TABLE IF EXISTS AcronymSF;

CREATE TABLE AcronymSF

(

     sf string -->pk

);

INSERT OVERWRITE TABLE AcronymSF

SELECT

    a[0].sf

FROM AcronymNonNormalized;

SELECT * FROM AcronymSF;

Figure 8. Acronym SF Contents

     For us to populate the rest of the tables, we will need to produce unique values for the arrays in these hierarchies that will be part of a composite primary key.   To do this we will take advantage of a Hive function called posexplode that will generate an auto integer for each array so we can uniquely identify them. The first table we want to create and populate like this will be the AcronymLFS table:

ADD JAR wasb://jymbo@jymbo.blob.core.windows.net/user/hdp/share/lib/hive/json-serde-1.1.9.3-SNAPSHOT-jar-with-dependencies.jar;

ADD JAR wasb://jymbo@jymbo.blob.core.windows.net/user/hdp/share/lib/hive/json-serde-1.1.9.3-SNAPSHOT.jar;

SET hive.execution.engine=tez;

SET hive.cli.print.header=true;

DROP TABLE IF EXISTS AcronymLFS;

CREATE TABLE AcronymLFS

(

     sf string, -->pk

     lfs int, -->pk

     lf string,

     freq int,

     since int

);

INSERT OVERWRITE TABLE AcronymLFS

SELECT

    a[0].sf,

    lfstable.seq AS lfs,

    lfs.lf,

    lfs.freq,

    lfs.since

FROM AcronymNonNormalized LATERAL VIEW POSEXPLODE( a[0].lfs) lfstable AS seq, lfs;

SELECT * FROM AcronymLFS;

Figure 9. AcronymLFS Contents

     

     To load the AcronymVARS table we need to use the same method we used for AcronymLFS, but one level down on the JSON hierarchy:

ADD JAR wasb://jymbo@jymbo.blob.core.windows.net/user/hdp/share/lib/hive/json-serde-1.1.9.3-SNAPSHOT-jar-with-dependencies.jar;

ADD JAR wasb://jymbo@jymbo.blob.core.windows.net/user/hdp/share/lib/hive/json-serde-1.1.9.3-SNAPSHOT.jar;

SET hive.execution.engine=tez;

SET hive.cli.print.header=true;

DROP TABLE IF EXISTS AcronymVARS;

CREATE TABLE AcronymVARS

(

     sf string, -->pk

     lfs int, -->pk

     vars int, -->pk

     lf string,

     freq int,

     since int

);

INSERT OVERWRITE TABLE AcronymVARS

SELECT

    a[0].sf,

    lfstable.seq as lfs,

    vartable.seq as vars,

    vars.lf,

    vars.freq,

    vars.since

FROM AcronymNonNormalized LATERAL VIEW POSEXPLODE( a[0].lfs) lfstable AS seq, lfs;

LATERAL VIEW POSEXPLODE(lfstable.lfs.vars) vartable AS seq, vars;

SELECT * FROM AcronymVARS;

Figure 10. AcronymVARS Content

     So with this we have normalized our JSON. If we execute a standard SQL query in Hive like this:

SET hive.execution.engine=tez;

SET hive.cli.print.header=true;

SELECT

 

     tsf.sf,

     tlfs.lf AS mainlf,

     tvars.lf,

     tvars.freq,

     tvars.since

      

FROM AcronymSF tsf

JOIN AcronymLFS tlfs ON tsf.sf=tlfs.sf

JOIN AcronymVARS tvars ON tlfs.sf=tvars.sf AND tlfs.lfs=tvars.lfs;

We get this:

Figure 11. SQL Join Result

     

     As you can see from these results, they match exactly the results from the query in figure 6. Now for performance purposes you would want to leave all this data in the same table(nested arrays and all). But if you wanted to set this data up to be moved to a relational database with sqoop, going through this technique may be useful.

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 { get; set; }

    public string fullname { get; set; }

    public string key { get; set; }

    public string url { get; set; }

}

//Alias of the author

public class Alias

{

    public string id { get; set; }

    public string name { get; set; }

    public string key { get; set; }

    public string url { get; set; }

}

//Authors of the paper

public class Author

{

    public Person person { get; set; }

    public Alias alias { get; set; }

}

//Root for Paper

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

}

#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